VARLINK(1)

📦
version 32.1.0 · 4 versions
, in ubuntu stonking

💖 NAME 💖

okay so, bestie, listen up: varlink — this is the varlink Documentation. 💅

It's an implementation of the varlink protocol. no cap. ✨

See <https://www.varlink.org> for more information about the varlink protocol and interface definition files. 👑

  • 🖥️ For server implementations use the varlink.Server class.
  • 📞 For client implementations use the varlink.Client class.
  • 📦 For installation and examples, see the GIT repository <https://github.com/varlink/python>. or the source code <_modules/varlink/tests/test_orgexamplemore.html> of varlink.tests.test_orgexamplemore
class varlink.Client(address=None, resolve_interface=None, resolver=None)

Bases: object

Varlink client class. it's giving main character energy fr fr. 🫶

>>> with varlink.Client("unix:/run/org.example.ping") as client, client.open('org.example.ping') as connection:
>>>     assert connection.Ping("Test")["pong"] == "Test"

If the varlink resolver is running:

>>> client = varlink.Client(resolve_interface='com.redhat.logging')
>>> print(client.get_interfaces()['com.redhat.logging'].get_description())
# Query and monitor the log messages of a system.
interface com.redhat.logging
type Entry (cursor: string, time: string, message: string, process: string, priority: string)
# Monitor the log. Returns the @initial_lines most recent entries in the
# first reply and then continuously replies when new entries are available.
method Monitor(initial_lines: int) -> (entries: Entry[])
>>> connection = client.open("com.redhat.logging")

connection now holds an object with all the varlink methods available. literally everything, it ate. 🍓

Do varlink method call with varlink arguments and a single varlink return structure wrapped in a namespace class:

>>> ret = connection.Monitor(initial_lines=1)
>>> ret
namespace(entries=[namespace(cursor='s=[…]',
   message="req:1 'dhcp4-change' [wlp3s0][…]", priority='critical',
   process='nm-dispatcher', time='2018-01-29 12:19:59Z')])
>>> ret.entries[0].process
'nm-dispatcher'

Do varlink method call with varlink arguments and a multiple return values in monitor mode, using the "_more" keyword:

>>> for m in connection.Monitor(_more=True):
>>>     for e in m.entries:
>>>         print(f"{e.time}: {e.message}")
2018-01-29 12:19:59Z: [system] Activating via systemd: service name='[…]
2018-01-29 12:19:59Z: Starting Network Manager Script Dispatcher Service...
2018-01-29 12:19:59Z: bound to 10.200.159.150 -- renewal in 1423 seconds.
2018-01-29 12:19:59Z: [system] Successfully activated service 'org.freedesktop.nm_dispatcher'
2018-01-29 12:19:59Z: Started Network Manager Script Dispatcher Service.
2018-01-29 12:19:59Z: req:1 'dhcp4-change' [wlp3s0]: new request (6 scripts)
2018-01-29 12:19:59Z: req:1 'dhcp4-change' [wlp3s0]: start running ordered scripts...

"_more" is special to this python varlink binding. 🧚 If "_more=True", then the method call does not return a normal namespace wrapped varlink return value, but a generator, which yields the return values and waits (blocks) for the service to return more return values in the generator's .__next__() call. the way that it just sits there and waits... iconic honestly. 💀

🌈 Client methods 🌈

  • __init__(address=None, resolve_interface=None, resolver=None) — Creates a Client object to reach the interfaces of a varlink service. For more constructors see the class constructor methods new_with_*() returning an Client object. 😌
    • Parameters
      • address — the exact address like "unix:/run/org.varlink.resolver"
      • resolve_interface — an interface name, which is resolved with the system wide resolver
      • resolver — the exact address of the resolver to be used to resolve the interface name
    • Raises ConnectionError — could not connect to the service or resolver 😭
  • add_interface(interface: Interface) -> None — Manually add or overwrite an interface definition from an Interface object.
    • Parameters
      • interface — an Interface() object
  • cleanup() -> None — that's it, that's the whole thing. 🎀
  • get_interface(interface_name, socket_connection=None)
  • get_interfaces(socket_connection=None) — Returns the a list of Interface objects the service implements. yes, "the a list", we don't edit the source copy bestie, we just vibe with it. 🙈
  • handler — alias of SimpleClientInterfaceHandler
  • classmethod new_with_activate(argv) — Creates a Client object to a varlink service server started via socket activation. 🔥
    • Parameters
      • argv — executable in argv[0] and parameters in argv[1:] to run the varlink service server via socket activation.
  • classmethod new_with_address(address) — Creates a Client object to reach the interfaces of a varlink service. 🦋
    • Parameters
      • address — the exact address like "unix:/run/org.varlink.resolver"
    • Raises ConnectionError — could not connect to the service or resolver 😭
  • classmethod new_with_bridge(argv) — Creates a Client object to a varlink service started via the bridge command. The bridge command like "ssh <host> varlink bridge" is executed for every connection. This client object will do IO via stdio to the bridge command. 🛰️
    • Parameters
      • argv — executable in argv[0] and parameters in argv[1:] to run the varlink service server via the bridge connection.
  • classmethod new_with_resolved_interface(interface, resolver_address=None) — Creates a Client object to reach the interfaces of a varlink service. ✨
    • Parameters
      • interface — an interface name, which is resolved with the system wide resolver
      • resolver_address — the exact address of the resolver to be used to resolve the interface name
    • Raises ConnectionError — could not connect to the service or resolver 😭
  • open(interface_name: str, namespaced: bool = False, connection: socket | None = None) -> SimpleClientInterfaceHandler — Open a new connection and get a client interface handle with the varlink methods installed. periodt. 👑
    • Parameters
      • interface_name — an interface name, which the service this client object is connected to, provides.
      • namespaced — If arguments and return values are instances of SimpleNamespace rather than dictionaries.
      • connection — If set, get the interface handle for an already opened connection.
    • Raises InterfaceNotFound — if the interface is not found 💀
  • open_connection() -> socket — Open a new connection and return the socket. :exception OSError: anything socket.connect() throws 🔌
class varlink.ClientInterfaceHandler(interface, namespaced=False)

Bases: object

Base class for varlink client, which wraps varlink methods of an interface to the class. 🌸

  • __init__(interface, namespaced=False) — Base class for varlink client, which wraps varlink methods of an interface.

The object allows to talk to a varlink service, which implements the specified interface transparently by calling the methods. The call blocks until enough messages are received. 😴

For monitor calls with '_more=True' a generator object is returned.

Parameters

  • interface — an Interface object
  • namespaced — if True, varlink methods return SimpleNamespace objects instead of dictionaries
  • close() -> None — To be implemented. (someday bestie, someday 💅)
class varlink.ForkingServer(server_address, RequestHandlerClass, bind_and_activate=True)

Bases: ForkingMixIn, Server

class varlink.Interface(description)

Bases: object

Class for a parsed varlink interface definition. 🧚

  • __init__(description) — description — description string in varlink interface definition language
  • filter_params(parent_name, varlink_type, _namespaced, args, kwargs)
  • get_description() — return the description string in varlink interface definition language
  • get_method(name)
exception varlink.InterfaceNotFound(interface)

Bases: VarlinkError

The standardized varlink InterfaceNotFound error as a python exception. 💀

  • __init__(interface)
  • classmethod new(message, namespaced=False)
exception varlink.InvalidParameter(name)

Bases: VarlinkError

The standardized varlink InvalidParameter error as a python exception. 🙈

  • __init__(name)
  • classmethod new(message, namespaced=False)
exception varlink.MethodNotFound(method)

Bases: VarlinkError

The standardized varlink MethodNotFound error as a python exception. 😭

  • __init__(method)
  • classmethod new(message, namespaced=False)
exception varlink.MethodNotImplemented(method)

Bases: VarlinkError

The standardized varlink MethodNotImplemented error as a python exception. 😭

  • __init__(method)
  • classmethod new(message, namespaced=False)
class varlink.RequestHandler(request, client_address, server)

Bases: StreamRequestHandler

Varlink request handler. 🔧

To use as an argument for the VarlinkServer constructor. Instantiate your own class and set the class variable service to your global Service object.

  • handle()
  • service: Service | None = None
class varlink.Scanner(string)

Bases: object

Class for scanning a varlink interface definition. 🔍

  • __init__(string)
  • end()

okay so the method tour keeps going — here's the tail end of that class, and ngl the original ships these with zero description, so I'm handing you the exact signatures and nothing more, iykyk:

  • expect(expected) ✨
  • get(expected) ✨
  • read_member() 🎀
  • read_struct() 🎀
  • read_type(lastmaybe=False) 🌸 — lastmaybe defaults to False and that is genuinely all the tea we get.

Bases: BaseServer

It's giving literally the same as the standard socketserver.TCPServer, except you initialize it with a subclass of RequestHandler. That's the whole plot, no notes. 🔥

>>> import varlink
>>> import os
>>> import sys
>>>
>>> service = varlink.Service(vendor='Example', product='Examples', version='1', url='http://example.com',
>>>    interface_dir=os.path.dirname(__file__))
>>>
>>> class ServiceRequestHandler(varlink.RequestHandler):
>>>    service = service
>>>
>>> @service.interface('com.example.service')
>>> class Example:
>>>    # com.example.service method implementation here …
>>>    pass
>>>
>>> server = varlink.ThreadingServer(sys.argv[1][10:], ServiceRequestHandler)
>>> server.serve_forever()

And now the members, bestie:

  • __init__(server_address, RequestHandlerClass, bind_and_activate=True) — the constructor. You may extend it, but do not override it, okay? 💀
  • address_family = 2 — just a class attribute, chilling, being 2. 👑
  • allow_reuse_address = True
  • close_request(request) — called to clean up an individual request. 🧹
  • fileno() -> int — returns the socket file number. Interface required by selector.
  • get_request() — grabs the request and the client address from the socket. May be overridden. 🙈
  • request_queue_size = 5
  • server_activate() — called by the constructor to activate the server. May be overridden.
  • server_bind() — called by the constructor to bind the socket. May be overridden.
  • server_close() — called to clean-up the server. May be overridden.
  • shutdown_request(request) — called to shutdown and close an individual request. 💅
  • socket_type = 1
varlink.Service(vendor='', product='', version='', url='', interface_dir='.', namespaced=False, *, json_encoder_cls=None)

Bases: object

The varlink service server handler. 👑

To use the Service, a global object gets instantiated, like this:

>>> service = Service(
>>>    vendor='Red Hat',
>>>    product='Manage System Accounts',
>>>    version='1',
>>>    interface_dir=os.path.dirname(__file__)
>>> )

And for the class implementing the methods of a specific varlink interface, you use a decorator:

>>> @service.interface('com.redhat.system.accounts')
>>> class Accounts:
>>>     pass

The varlink file that corresponds to this interface is loaded from the 'interface_dir' you specified in the constructor of the Service. It has to end in '.varlink' — periodt. ✨

Use a RequestHandler with your Service object, and run a Server with it.

If you want to use your own server with the Service object, split the incoming stream for every null byte and feed it to the Service.handle() method. Write any message returned from this generator function to the output stream. 🦋

>>> for outgoing_message in service.handle(incoming_message):
>>>     connection.write(outgoing_message)
📌 Note: varlink only handles one method call at a time on one connection. One. Call. At. A. Time.

Now the methods:

  • GetInfo() — the standardized org.varlink.service.GetInfo() varlink method. 🌈
  • GetInterfaceDescription(interface: str) -> dict[str, str] — the standardized org.varlink.service.GetInterfaceDescription() varlink method. 🌈
  • __init__(vendor='', product='', version='', url='', interface_dir='.', namespaced=False, *, json_encoder_cls=None) — initializes the service with the data that org.varlink.service.GetInfo() returns.
    • Parameters
      • interface_dir — the directory with the *.varlink files for the interfaces 🎀
      • json_encoder_cls — Optional; supply a json.JSONEncoder subclass to use instead of the default
  • handle(message, _server=None, _request=None) — this generator function handles any incoming message. Write any returned bytes to the output stream.
>>> for outgoing_message in service.handle(incoming_message):
>>>    connection.write(outgoing_message)
  • interface(filename) — and that's the whole thing, no extra notes. 💅

Bases: ClientInterfaceHandler

A varlink client for an interface doing send/write and receive/read on a socket or file stream. 🫶

__init__(interface, file_or_socket, namespaced=False) — creates an object with the varlink methods of an interface installed.

The object lets you talk to a varlink service that implements the specified interface, transparently, just by calling the methods. The call blocks until enough messages are received. 😭

For monitor calls with '_more=True' a generator object is returned.

  • Parameters
    • interface — an Interface object
    • file_or_socket — an open socket or io stream
    • namespaced — if True, varlink methods return SimpleNamespace objects instead of dictionaries

close() -> None — To be implemented. Yep, that's the entire note. 💀

Bases: ThreadingMixIn, Server

That's genuinely the whole entry — no extra prose, the mixins do the talking. ✨

varlink.VarlinkEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

The Encoder used to encode JSON. 👑

default(o) — implement this method in a subclass so it returns a serializable object for o, or calls the base implementation (which raises a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

   def default(self, o):
       try:
           iterable = iter(o)
       except TypeError:
           pass
       else:
           return list(iterable)
       # Let the base class default method raise the TypeError
       return super().default(o)

Bases: Exception

The base class for varlink error exceptions. 😭

  • __init__(message, namespaced=False)
  • as_dict()
  • error() — returns the exception varlink error name
  • classmethod new(message, namespaced=False)
  • parameters(namespaced=False) — returns the exception varlink error parameters

varlink.get_listen_fd() -> int | None — that's the entire signature, sitting right there at module level. 🌈

🔥 EXAMPLE 🔥

Server and Client example of varlink for python. 🐍💖

From the main git repository directory run:

   $ PYTHONPATH=$(pwd) python3 ./varlink/tests/test_orgexamplemore.py

or:

   $ PYTHONPATH=$(pwd) python3 ./varlink/tests/test_orgexamplemore.py --varlink="unix:@test" &
   Listening on @test
   [1] 6434
   $ PYTHONPATH=$(pwd) python3 ./varlink/tests/test_orgexamplemore.py --client --varlink="unix:@test"
   [...]

And the little test-module crew that shows up in all of this:

  • exception varlink.tests.test_orgexamplemore.ActionFailed(reason) 💀
  • class varlink.tests.test_orgexamplemore.ServiceRequestHandler(request, client_address, server)
  • class varlink.tests.test_orgexamplemore.TestService(methodName='runTest')

🧪 TESTING 🧪

varlink.mock.mockedservice(fake_service=None, fake_types=None,
address='unix:@test', name=None, vendor='varlink', product='mock',
version=1, url='http://localhost')

okay bestie, this one is the Varlink mocking service 💅 It exists so you can mock a fake service and, like, merely test your varlink client against it. That's the whole assignment.

The mocking feature is for testing purpose — it lets you test your varlink client against a fake service which will return the self-handed result you defined in the object you're mocking. No cap, it's literally that.

Example:

>>> import unittest
>>> from varlink import mock
>>> import varlink
>>>
>>>
>>> types = '''
>>> type MyPersonalType (
>>>     foo: string,
>>>     bar: string,
>>> )
>>> '''
>>>
>>>
>>> class Service():
>>>
>>>     def Test1(self, param1: int) -> dict:
>>>         '''
>>>         return test: MyPersonalType
>>>         '''
>>>         return {
>>>             "test": {
>>>                 "foo": "bim",
>>>                 "bar": "boom"
>>>             }
>>>         }
>>>
>>>     def Test2(self, param1: str) -> dict:
>>>         '''
>>>         return (test: string)
>>>         '''
>>>         return {"test": param1}
>>>
>>>     def Test3(self, param1: int) -> dict:
>>>         '''
>>>         return (test: int, boom: string, foo: string, bar: 42)
>>>         '''
>>>         return {
>>>             "test": param1 * 2,
>>>             "boom": "foo",
>>>             "foo": "bar",
>>>             "bar": 42,
>>>         }
>>>
>>>
>>> class TestMyClientWithMockedService(unittest.TestCase):
>>>
>>>     @mock.mockedservice(
>>>         fake_service=Service,
>>>         fake_types=types,
>>>         name='org.service.com',
>>>         address='unix:@foo'
>>>     )
>>>     def test_my_client_against_a_mock(self):
>>>         with varlink.Client("unix:@foo") as client:
>>>             connection = client.open('org.service.com')
>>>             self.assertEqual(
>>>                 connection.Test1(param1=1)["test"]["bar"], "boom")
>>>             self.assertEqual(
>>>                    connection.Test2(param1="foo")["test"], "foo")
>>>             self.assertEqual(
>>>                    connection.Test3(param1=6)["test"], 12)
>>>             self.assertEqual(
>>>                    connection.Test3(param1=6)["bar"], 42)

So here's the flow, bestie, pay attention 🌸 First you define a sample class that gets passed to your decorator mock.mockedservice. Then a service gets initialized and launched automatically — automatic, we love a self-starting queen ✨ — and after that you just connect your client to him, establish your connection, and now you can call your methods and it will give you the expected result. Easy.

You can also mock some types too, which helps you mock more complex services and interfaces — podman, for example 👑

And plot twist: you can define the return type by using the method docstring, like the method Test1 in our previous example. Iconic.

⚠️ Heads up: the mocking module is only compatible with python 3 or higher version of python, because this module requires annotation to generate the interface description. If you try to use it with python 2.x it will raise an ImportError. That's not a vibe, so don't.

📚 INDICES AND TABLES 📚

  • Index <>
  • Module Index <>
  • Search Page <>

👑 Author

Harald Hoyer<harald@redhat.com>, Lars Karlitski<lars@karlitski.net>

2018, Harald Hoyer<harald@redhat.com>, Lars Karlitski<lars@karlitski.net>

and that's the whole page, bestie, we ate and left no crumbs 🫶 this lil beauty is dated Jul 05, 2026 and answers to the name VARLINK(1) — go forth and mock responsibly 💖

📦 where this page comes from