Provided by: python3-varlink_31.0.0-3_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
2024, Harald Hoyer<harald@redhat.com>, Lars Karlitski<lars@karlitski.net>
Apr 26, 2024 VARLINK(1)