varlink for python

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.

class varlink.SimpleClientInterfaceHandler(interface, file_or_socket, namespaced=False)

Bases: varlink.client.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.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.RequestHandler(request, client_address, server)

Bases: socketserver.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.Server(server_address, RequestHandlerClass, bind_and_activate=True)

Bases: socketserver.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.ThreadingServer(server_address, RequestHandlerClass, bind_and_activate=True)

Bases: socketserver.ThreadingMixIn, varlink.server.Server

class varlink.ForkingServer(server_address, RequestHandlerClass, bind_and_activate=True)

Bases: socketserver.ForkingMixIn, varlink.server.Server

exception varlink.InterfaceNotFound(interface)

Bases: varlink.error.VarlinkError

The standardized varlink InterfaceNotFound error as a python exception

__init__(interface)

Initialize self. See help(type(self)) for accurate signature.

classmethod new(message, namespaced=False)

exception varlink.MethodNotFound(method)

Bases: varlink.error.VarlinkError

The standardized varlink MethodNotFound error as a python exception

__init__(method)

Initialize self. See help(type(self)) for accurate signature.

classmethod new(message, namespaced=False)

exception varlink.MethodNotImplemented(method)

Bases: varlink.error.VarlinkError

The standardized varlink MethodNotImplemented error as a python exception

__init__(method)

Initialize self. See help(type(self)) for accurate signature.

classmethod new(message, namespaced=False)

exception varlink.InvalidParameter(name)

Bases: varlink.error.VarlinkError

The standardized varlink InvalidParameter error as a python exception

__init__(name)

Initialize self. See help(type(self)) for accurate signature.

classmethod new(message, namespaced=False)

exception varlink.ConnectionError

Bases: OSError

Connection error.

__init__()

Initialize self. See help(type(self)) for accurate signature.

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

Bases: json.encoder.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)

Initialize self. See help(type(self)) for accurate signature.

as_dict()

error()

returns the exception varlink error name

classmethod new(message, namespaced=False)

parameters(namespaced=False)

returns the exception varlink error parameters

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)

class varlink.Scanner(string)

Bases: object

Class for scanning a varlink interface definition.

__init__(string)

Initialize self. See help(type(self)) for accurate signature.

end()

expect(expected)

get(expected)

read_member()

read_struct()

read_type(lastmaybe=False)

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