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.

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

Connection error.

Class for a parsed varlink interface definition.

The standardized varlink InterfaceNotFound error as a python exception

The standardized varlink InvalidParameter error as a python exception

The standardized varlink MethodNotFound error as a python exception

The standardized varlink MethodNotImplemented error as a python exception

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.

Class for scanning a varlink interface definition.

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()
    

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.

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

The Encoder used to encode JSON

The base class for varlink error exceptions

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.