Provided by: python3-varlink_31.0.0-4_all 
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 super().default(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
2018, Harald Hoyer<harald@redhat.com>, Lars Karlitski<lars@karlitski.net>
Feb 16, 2025 VARLINK(1)