VARLINK(1)

NAME
varlink - varlink Documentation

An implementation of the varlink protocol

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 of varlink.tests.test_orgexamplemore

    class varlink.Client(address=None, resolve_interface=None, resolver=None)
        Bases: object

        Varlink client class.

            >>> 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.

        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("%s: %s" % (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.

        __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)
            Manually add or overwrite an interface definition from an Interface object.

            Parameters
                interface -- an Interface() object

        cleanup()

        get_interface(interface_name, socket_connection=None)

        get_interfaces(socket_connection=None)
            Returns the a list of Interface objects the service implements.

        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, namespaced=False, connection=None)
            Open a new connection and get a client interface handle with the varlink methods installed.

            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()
            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()
            To be implemented.

    exception varlink.ConnectionError
        Bases: OSError

        Connection error.

        __init__(*args, **kwargs)

    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 = None

    class varlink.Scanner(string)
        Bases: object

        Class for scanning a varlink interface definition.

        __init__(string)

        end()

        expect(expected)

        get(expected)

        read_member()

        read_struct()

        read_type(lastmaybe=False)

    class varlink.Server(server_address, RequestHandlerClass, bind_and_activate=True)
        Bases: BaseServer

        The same as the standard socketserver.TCPServer, to initialize with a subclass of RequestHandler.

            >>> import varlink
            >>> import os
            >>>
            >>> 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()

        __init__(server_address, RequestHandlerClass, bind_and_activate=True)
            Constructor. May be extended, do not override.

        address_family = 2

        allow_reuse_address = True

        close_request(request)
            Called to clean up an individual request.

        fileno()
            Return socket file number.

            Interface required by selector.

        get_request()
            Get the request and client address from the socket.

            May be overridden.

        request_queue_size = 5

        server_activate()
            Called by constructor to activate the server.

            May be overridden.

        server_bind()
            Called by 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

    class varlink.Service(vendor='', product='', version='', url='', interface_dir='.', namespaced=False)
        Bases: object

        Varlink service server handler

        To use the Service, a global object is instantiated:

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

        For the class implementing the methods of a specific varlink interface a decorator is used:

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

        The varlink file corresponding to this interface is loaded from the 'interface_dir' specified in the constructor of the Service. It has to end in '.varlink'.

        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.

        GetInfo()
            The standardized org.varlink.service.GetInfo() varlink method.

        GetInterfaceDescription(interface)
            The standardized org.varlink.service.GetInterfaceDescription() varlink method.

        __init__(vendor='', product='', version='', url='', interface_dir='.', namespaced=False)
            Initialize the service with the data org.varlink.service.GetInfo() returns

            Parameters
                interface_dir -- the directory with the *.varlink files for the interfaces

        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)

    class varlink.SimpleClientInterfaceHandler(interface, file_or_socket, namespaced=False)
        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 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

                •   file_or_socket -- an open socket or io stream

                •   namespaced -- if True, varlink methods return SimpleNamespace objects instead of dictionaries

        close()
            To be implemented.

    class varlink.ThreadingServer(server_address, RequestHandlerClass, bind_and_activate=True)
        Bases: ThreadingMixIn, Server

    class 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 such that it returns a serializable object for o, or calls the base implementation (to raise 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 JSONEncoder.default(self, o)

    exception varlink.VarlinkError(message, namespaced=False)
        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()

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"
            [...]

    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')
        Varlink mocking service

        To mock a fake service and merely test your varlink client against.

        The mocking feature is for testing purpose, it's allow you to test your varlink client against a fake service which will returned self handed result defined in your object who will be mocked.

        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)

        First you need to define a sample class that will be passed to your decorator mock.mockedservice and then a service will be initialized and launched automatically, and after that you just need to connect your client to him and to establish your connection then now you can call your methods and it will give you the expected result.

        You can also mock some types too, to help you to mock more complex service and interfaces like podman by example.

        You can define the return type by using the method docstring like the method Test1 in our previous example.

        The mocking module is only compatible with python 3 or higher version of python because this module require annotation to generate interface description.

        If you try to use it with python 2.x it will raise an ImportError.

INDICES AND TABLES

    •   Index

    •   Module Index

    •   Search Page

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

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