| PYTHON-SDBUS(1) | python-sdbus | PYTHON-SDBUS(1) |
python-sdbus - python-sdbus
Python-sdbus is the python D-Bus library that aim to use the modern features of python
D-Bus is the inter-process communication standard commonly used on Linux desktop.
This documentation expects you to be familiar with D-Bus concepts and conventions.
If you are unfamiliar with D-Bus you might want to read following pages:
Wikipedia page
Lennart Poettering post about D-Bus
D-Bus specification by freedesktop.org
Install D-Feet D-Bus debugger and observe services and objects on your D-Bus
Python-sdbus supports both blocking and async IO.
Regular python functions are always blocking.
Asyncio is a part of python standard library that allows non-blocking io.
Asyncio documentation
Generally blocking IO should only be used for simple scripts and programs that interact with existing D-Bus objects.
Blocking quick start
Blocking API
Asyncio quick start
Asyncio API
D-Bus types reference
NOTE:
Unsigned integers range is 0 < (2**bit_size)-1.
Signed integers range is -(2**(bit_size-1)) < (2**(bit_size-1))-1.
| Name | D-Bus type | Python type | Description |
| Boolean | b | bool | True or False |
| Byte | y | int | Unsigned 8-bit integer. Note: array of bytes (ay) has different type in python domain. |
| Int16 | n | int | Signed 16-bit integer. |
| Uint16 | q | int | Unsigned 16-bit integer. |
| Int32 | i | int | Signed 32-bit integer. |
| Uint32 | u | int | Unsigned 32-bit integer. |
| Int64 | x | int | Signed 64-bit integer. |
| Uint64 | t | int | Unsigned 64-bit integer. |
| Double | d | float | Float point number |
| Unix FD | h | int | File descriptor |
| String | s | str | String |
| Object Path | o | str | Syntactically correct D-Bus object path |
| Signature | g | str | D-Bus type signature |
| Array | a | list | List of some single type. Example: as array of strings |
| Byte Array | ay | bytes | Array of bytes. Not a unique type in D-Bus but a different type in Python. Accepts both bytes and bytearray. Used for binary data. |
| Struct | () | tuple | Tuple. Example: (isax) tuple of int, string and array of int. |
| Dictionary | a{} | dict | Dictionary with key type and value type. Note: Dictionary is always a part of array. I.E. a{si} is the dict with string keys and integer values. {si} is NOT a valid signature. |
| Variant | v | tuple | Unknown type that can be any single type. In Python represented by a tuple of a signature string and a single type. Example: ("s", "test") variant of a single string |
D-Bus uses CamelCase for method names.
Python uses snake_case.
When decorating a method name will be automatically translated from snake_case to CamelCase. Example: close_notification -> CloseNotification
However, all decorators have a parameter to force D-Bus name to a specific value. See API documentation for a particular decorator.
Most object methods that take a bus as a parameter will use a thread-local default bus connection if a bus object is not explicitly passed.
Session bus is default bus when running as a user and system bus otherwise.
request_default_bus_name_async() can be used to acquire a service name on default bus.
Use sd_bus_open_user() and sd_bus_open_system() to acquire a specific bus connection.
Set the default connection to a new default with set_default_bus(). This should be done before any object that take bus as an init argument are created.
In the future there will be a better way to create and acquire new bus connections.
These calls are shared between async and blocking API.
Should be called before you create any objects that might use default bus.
Default bus can be replaced but the change will only affect newly created objects.
Example on how systemd encodes unit names on D-Bus:
from sdbus import encode_object_path
# System uses /org/freedesktop/systemd1/unit as prefix of all units
# dbus.service is a name of D-Bus unit but dot . is not a valid object path
s = encode_object_path('/org/freedesktop/systemd1/unit', 'dbus.service')
print(s)
# Prints: /org/freedesktop/systemd1/unit/dbus_2eservice
Example decoding systemd unit name:
from sdbus import decode_object_path
s = decode_object_path(
'/org/freedesktop/systemd1/unit',
'/org/freedesktop/systemd1/unit/dbus_2eservice'
)
print(s)
# Prints: dbus.service
Flags are int values that should be ORed to combine.
Example, DbusDeprecatedFlag plus DbusHiddenFlag: DbusDeprecatedFlag | DbusHiddenFlag
Python-sdbus works by declaring interface classes.
Interface classes for blocking IO should be derived from DbusInterfaceCommon.
The class constructor takes interface_name keyword to determine the D-Bus interface name for all D-Bus elements declared in the class body.
Example:
class ExampleInterface(DbusInterfaceCommon,
interface_name='org.example.myinterface'
):
...
Interface class body should contain the definitions of methods and properties using the decorators dbus_method() and dbus_property() respectively.
Example:
from sdbus import (DbusInterfaceCommon,
dbus_method, dbus_property)
class ExampleInterface(DbusInterfaceCommon,
interface_name='org.example.myinterface'
):
# Method that takes an integer and does not return anything
@dbus_method('u')
def close_notification(self, an_int: int) -> None:
raise NotImplementedError
# Read only property of int
@dbus_property()
def test_int(self) -> int:
raise NotImplementedError
This is an interface of that defines a one D-Bus method and one property.
The actual body of the decorated function will not be called. Instead the call will be routed through D-Bus to a another process. Interface can have non-decorated functions that will act as regular methods.
Blocking IO can only interact with existing D-Bus objects and can not be served for other processes to interact with. See Blocking vs Async
DbusInterfaceCommon.__init__() method takes service_name and object_path of the remote object that the object will proxy to.
Example creating a proxy and calling method:
...
# Initialize the object
d = ExampleInterface(
service_name='org.example.test',
object_path='/',
)
d.close_notification(1234)
NOTE:
Methods are functions wrapped with dbus_method() decorator.
If the remote object sends an error reply an exception with base of DbusFailedError will be raised. See Exceptions for list of exceptions.
The wrapped function will not be called. Its recommended to set the function to raise NotImplementedError.
Example:
from sdbus import DbusInterfaceCommon, dbus_method
class ExampleInterface(...):
...
# Body of some class
@dbus_method('u')
def close_notification(self, an_int: int) -> None:
raise NotImplementedError
D-Bus property is defined by wrapping a function with dbus_property() decorator.
Example:
from sdbus import DbusInterfaceCommon, dbus_property
class ExampleInterface(...):
...
# Body of some class
# Property of str
@dbus_property('s')
def test_string(self) -> str:
raise NotImplementedError
The new property behaves very similar to Pythons property() decorator.
# Initialize the proxy
d = ExampleInterface(
service_name='org.example.test',
object_path='/',
)
# Print it
print(d.test_string)
# Assign new string
d.test_string = 'some_string'
If property is read-only when DbusPropertyReadOnlyError will be raised.
A D-Bus object can have multiple interfaces with different methods and properties.
To implement this define multiple interface classes and do a multiple inheritance on all interfaces the object has.
Example:
from sdbus import DbusInterfaceCommon, dbus_method
class ExampleInterface(DbusInterfaceCommon,
interface_name='org.example.myinterface'
):
@dbus_method('i')
def example_method(self, an_int: int) -> None:
raise NotImplementedError
class TestInterface(DbusInterfaceCommon,
interface_name='org.example.test'
):
@dbus_method('as')
def test_method(self, str_array: List[str]) -> None:
raise NotImplementedError
class MultipleInterfaces(TestInterface, ExampleInterface):
...
MultipleInterfaces class will have both test_method and example_method that will be proxied to correct interface names. (org.example.myinterface and org.example.test respectively)
Useful to test if connection or remote service is alive.
WARNING:
It is users responsibility to parse that data.
Equivalent to GetAll method of the org.freedesktop.DBus.Properties interface but the member names are automatically translated to python names. (internally calls it for each interface used in class definition)
Example:
from sdbus import (DbusInterfaceCommon,
dbus_method, dbus_property)
class ExampleInterface(DbusInterfaceCommon,
interface_name='org.example.my'
):
# Method that takes an integer and does not return anything
@dbus_method('u')
def close_notification(self, an_int: int) -> None:
raise NotImplementedError
# Method that does not take any arguments and returns a list of str
@dbus_method()
def get_capabilities(self) -> List[str]:
raise NotImplementedError
# Method that takes a dict of {str: str} and returns an int
@dbus_method('a{ss}')
def count_entries(self, a_dict: Dict[str, str]) -> int:
raise NotImplementedError
# Read only property of int
@dbus_property()
def test_int(self) -> int:
raise NotImplementedError
# Read/Write property of str
@dbus_property('s')
def test_string(self) -> str:
raise NotImplementedError
Dict[ObjectPath, Dict[InterfaceName, Dict[PropertyName, PropertyValue]]]
Decorated function becomes linked to D-Bus method. Always use round brackets () even when not passing any arguments.
modifies behavior. No effect on remote connections. Defaults to 0 meaning no special behavior.
See Flags .
Defining methods example:
from sdbus import DbusInterfaceCommon, dbus_method
class ExampleInterface(DbusInterfaceCommon,
interface_name='org.example.my'
):
# Method that takes an integer and does not return anything
@dbus_method('u')
def close_notification(self, an_int: int) -> None:
raise NotImplementedError
# Method that does not take any arguments and returns a list of str
@dbus_method()
def get_capabilities(self) -> List[str]:
raise NotImplementedError
# Method that takes a dict of {str: str} and returns an int
@dbus_method('a{ss}')
def count_entries(self, a_dict: Dict[str, str]) -> int:
raise NotImplementedError
Calling methods example:
# Initialize the object
d = ExampleInterface(
service_name='org.example.test',
object_path='/',
)
d.close_notification(1234)
l = d.get_capabilities()
d.count_entries({'a': 'asdasdasd', 'b': 'hgterghead213d'})
Property works just like @property decorator would. Always use round brackets () even when not passing any arguments.
Read only property can be indicated by passing empty D-Bus signature "".
Trying to assign a read only property will raise AttributeError
modifies behavior. No effect on remote connections. Defaults to 0 meaning no special behavior.
See Flags .
Defining properties example:
from sdbus import DbusInterfaceCommon, dbus_property
class ExampleInterface(DbusInterfaceCommon,
interface_name='org.example.myproperty'
):
# Property of int
@dbus_property('i')
def test_int(self) -> int:
raise NotImplementedError
# Property of str
@dbus_property('s')
def test_string(self) -> str:
raise NotImplementedError
Properties usage example:
# Initialize the object
d = ExampleInterface(
service_name='org.example.test',
object_path='/',
)
# Print the int
print(d.test_int)
# Assign new string
d.test_string = 'some_string'
# Print it
print(d.test_string)
Python-sdbus works by declaring interface classes.
Interface classes for async IO should be derived from DbusInterfaceCommonAsync.
The class constructor takes interface_name keyword to determine the D-Bus interface name for all D-Bus elements declared in the class body.
Example:
from sdbus import DbusInterfaceCommonAsync
class ExampleInterface(DbusInterfaceCommonAsync,
interface_name='org.example.myinterface'
):
...
Interface class body should contain the definitions of methods, properties and signals using decorators such as dbus_method_async(), dbus_property_async() and dbus_signal_async().
Example:
from sdbus import (DbusInterfaceCommonAsync, dbus_method_async,
dbus_property_async, dbus_signal_async)
class ExampleInterface(DbusInterfaceCommonAsync,
interface_name='org.example.myinterface'
):
# Method that takes an integer and multiplies it by 2
@dbus_method_async('i', 'i')
async def double_int(self, an_int: int) -> None:
return an_int * 2
# Read only property of str
@dbus_property_async('s')
def read_string(self) -> int:
return 'Test'
# Signal with a list of strings
@dbus_signal_async('as')
def str_signal(self) -> List[str]:
raise NotImplementedError
DbusInterfaceCommonAsync provides two methods for proxying remote objects.
DbusInterfaceCommonAsync.new_proxy() class method bypasses the class __init__ and returns proxy object.
DbusInterfaceCommonAsync._proxify() should be used inside the __init__ methods if your class is a proxy only.
Recommended to create proxy classes that a subclass of the interface:
from sdbus import DbusInterfaceCommonAsync
class ExampleInterface(...):
# Some interface class
...
class ExampleClient(ExampleInterface):
def __init__(self) -> None:
# Your client init can proxy to any object based on passed arguments.
self._proxify('org.example.test', '/')
NOTE:
DbusInterfaceCommonAsync.export_to_dbus() method will export the object to the D-Bus. After calling it the object becomes visible on D-Bus for other processes to call.
Example using ExampleInterface from before:
from sdbus import request_default_bus_name_async
loop = get_event_loop()
i = ExampleInterface()
async def start() -> None:
# Acquire a name on the bus
await request_default_bus_name_async('org.example.test')
# Start serving at / path
i.export_to_dbus('/')
loop.run_until_complete(start())
loop.run_forever()
The interface objects are designed to be transparent to their connection status. This means if the object not proxied to remote the calls to decorated methods will still work in the local scope.
This is the call to local object:
i = ExampleInterface()
async def test() -> None:
print(await i.double_int(5)) # Will print 10
This is a call to remote object at 'org.example.test' service name and '/' path:
i = ExampleInterface.new_proxy('org.example.test', '/')
async def test() -> None:
print(await i.double_int(5)) # Will print 10
Methods are async function calls wrapped with dbus_method_async() decorator. (see the API reference for decorator parameters)
Methods have to be async function, otherwise AssertionError will be raised.
While method calls are async there is a inherit timeout timer for any method call.
To return an error to caller you need to raise exception which has a DbusFailedError as base. Regular exceptions will not propagate.
See Exceptions.
Example:
from sdbus import DbusInterfaceCommonAsync, dbus_method_async
class ExampleInterface(...):
...
# Body of some class
# Method that takes a string
# and returns uppercase of that string
@dbus_method_async(
input_signature='s',
result_signature='s',
result_args_names=('uppercased', ) # This is optional but
# makes arguments have names in
# instrospection data.
)
async def upper(self, str_to_up: str) -> str:
return str_to_up.upper()
Methods behave exact same way as Python methods would:
print(await example_object.upper('test')) # prints TEST
Properties are a single value that can be read and write.
To declare a read only property you need to decorate a regular function with dbus_property_async() decorator.
Example:
from sdbus import DbusInterfaceCommonAsync, dbus_property_async
class ExampleInterface(...):
...
# Body of some class
# Read only property. No setter defined.
@dbus_property_async('i')
def read_only_number(self) -> int:
return 10
To create a read/write property you need to decorate the setter function with the setter attribute of your getter function.
Example:
from sdbus import DbusInterfaceCommonAsync, dbus_property_async
class ExampleInterface(...):
...
# Body of some class
# Read/write property. First define getter.
@dbus_property_async('s')
def read_write_str(self) -> str:
return self.s
# Now create setter. Method name does not matter.
@read_write_str.setter # Use the property setter method as decorator
def read_write_str_setter(self, new_str: str) -> None:
self.s = new_str
Properties are supposed to be lightweight. Make sure you don't block event loop with getter or setter.
Async properties do not behave the same way as property() decorator does.
To get the value of the property you can either directly await on property or use get_async() method. (also need to be awaited)
To set property use set_async() method.
Example:
...
# Somewhere in async function
# Assume we have example_object of class defined above
print(await example_object.read_write_str) # Print the value of read_write_str
...
# Set read_write_str to new value
await example_object.read_write_str.set_async('test')
To define a D-Bus signal wrap a function with dbus_signal_async() decorator.
The function is only used for type hints information. It is recommended to just put raise NotImplementedError in to the body of the function.
Example:
from sdbus import DbusInterfaceCommonAsync, dbus_signal_async
class ExampleInterface(...):
...
# Body of some class
@dbus_signal_async('s')
def name_changed(self) -> str:
raise NotImplementedError
To catch a signal use async for loop:
async for x in example_object.name_changed:
print(x)
WARNING:
A signal can be emitted with emit() method.
Example:
example_object.name_changed.emit('test')
Signals can also be caught from multiple D-Bus objects using catch_anywhere() method. The async iterator will yield the path of the object that emitted the signal and the signal data.
catch_anywhere() can be called from class but in such case the service name must be provided.
Example:
async for path, x in ExampleInterface.name_changed('org.example.test'):
print(f"On {path} caught: {x}")
If you define a subclass which overrides a declared D-Bus method or property you need to use dbus_method_async_override() and dbus_property_async_override() decorators. Overridden property can decorate a new setter.
Overridden methods should take same number and type of arguments.
Example:
from sdbus import (dbus_method_async_override,
dbus_property_async_override)
# Some subclass
class SubclassInterface(...):
...
@dbus_method_async_override()
async def upper(self, str_to_up: str) -> str:
return 'Upper: ' + str_to_up.upper()
@dbus_property_async_override()
def str_prop(self) -> str:
return 'Test property' + self.s
# Setter needs to be decorated again to override
@str_prop.setter
def str_prop_setter(self, new_s: str) -> None:
self.s = new_s.upper()
A D-Bus object can have multiple interfaces with different methods and properties.
To implement this define multiple interface classes and do a multiple inheritance on all interfaces the object has.
Example:
from sdbus import DbusInterfaceCommonAsync
class ExampleInterface(DbusInterfaceCommonAsync,
interface_name='org.example.myinterface'
):
@dbus_method_async('i', 'i')
async def double_int(self, an_int: int) -> None:
return an_int * 2
class TestInterface(DbusInterfaceCommonAsync,
interface_name='org.example.test'
):
@dbus_method_async('as', 's')
async def join_str(self, str_array: List[str]) -> str:
return ''.join(str_array)
class MultipleInterfaces(TestInterface, ExampleInterface):
...
MultipleInterfaces class will have both test_method and example_method that will be wired to correct interface names. (org.example.myinterface and org.example.test respectively)
NOTE:
Useful to test if connection or remote service is alive.
WARNING:
It is users responsibility to parse that data.
Equivalent to GetAll method of the org.freedesktop.DBus.Properties interface but the member names are automatically translated to python names. (internally calls it for each interface used in class definition)
sdbus.utils.parse_properties_changed() can be used to transform this signal data in to an easier to work with dictionary.
Signal data is:
Example of serving objects with ObjectManager:
my_object_manager = DbusObjectManagerInterfaceAsync()
my_object_manager.export_to_dbus('/object/manager')
managed_object = DbusInterfaceCommonAsync()
my_object_manager.export_with_manager('/object/manager/example')
Dict[ObjectPath, Dict[InterfaceName, Dict[PropertyName, PropertyValue]]]
sdbus.utils.parse_interfaces_added() can be used to make signal data easier to work with.
Signal data is:
sdbus.utils.parse_interfaces_removed() can be used to make signal data easier to work with.
Signal data is:
ObjectManager must be exported first.
Path should be a subpath of where ObjectManager was exported. Example, if ObjectManager exported to /object/manager, the managed object can be exported at /object/manager/test.
ObjectManager will keep the reference to the object.
Releases reference to the object.
CAUTION:
Underlying function must be a coroutine function.
modifies behavior. No effect on remote connections. Defaults to 0 meaning no special behavior.
See Flags .
sequence of result argument names.
These names will show up in introspection data but otherwise have no effect.
Sequence can be list, tuple, etc... Number of elements in the sequence should match the number of result arguments otherwise SdBusLibraryError will be raised.
Defaults to result arguments being nameless.
sequence of input argument names.
These names will show up in introspection data but otherwise have no effect.
Sequence can be list, tuple, etc... Number of elements in the sequence should match the number of result arguments otherwise RuntimeError will be raised.
If result_args_names has been passed when Python function argument names will be used otherwise input arguments will be nameless
Example:
from sdbus import DbusInterfaceCommonAsync, dbus_method_async
class ExampleInterface(DbusInterfaceCommonAsync,
interface_name='org.example.test'
):
# Method that takes a string
# and returns uppercase of that string
@dbus_method_async(
input_signature='s',
result_signature='s',
result_args_names=('uppercased', ) # This is optional but
# makes arguments have names in
# instrospection data.
)
async def upper(self, str_to_up: str) -> str:
return str_to_up.upper()
The underlying function has to be a regular def function.
The property will be read-only or read/write based on if setter was declared.
WARNING:
modifies behavior. No effect on remote connections. Defaults to 0 meaning no special behavior.
See Flags .
Properties have following methods:
See example on how to use.
Calling the setter locally will emit properties_changed signal to D-Bus.
The property can also be directly await ed instead of calling this method.
Example:
from sdbus import DbusInterfaceCommonAsync, dbus_property_async
class ExampleInterface(DbusInterfaceCommonAsync,
interface_name='org.example.test'
):
def __init__(self) -> None:
# This is just a generic init
self.i = 12345
self.s = 'test'
# Read only property. No setter defined.
@dbus_property_async('i')
def read_only_number(self) -> int:
return self.i
# Read/write property. First define getter.
@dbus_property_async('s')
def read_write_str(self) -> str:
return self.s
# Now create setter. Method name does not matter.
@read_write_str.setter # Use the property setter method as decorator
def read_write_str_setter(self, new_str: str) -> None:
self.s = new_str
Underlying function return type hint is used for signal type hints.
sequence of signal argument names.
These names will show up in introspection data but otherwise have no effect.
Sequence can be list, tuple, etc... Number of elements in the sequence should match the number of result arguments otherwise RuntimeError will be raised.
Defaults to result arguments being nameless.
modifies behavior. No effect on remote connections. Defaults to 0 meaning no special behavior.
See Flags .
Signals have following methods:
This is main way to await for new events.
Both remote and local objects operate the same way.
Signal objects can also be async iterated directly: async for x in something.some_signal
async for path, data in something.some_signal.catch_anywhere():
This method can be called from both an proxy object and class. However, it cannot be called on local objects and will raise NotImplementedError.
Example:
from sdbus import DbusInterfaceCommonAsync, dbus_signal_async
class ExampleInterface(DbusInterfaceCommonAsync,
interface_name='org.example.signal'
):
@dbus_signal_async('s')
def name_changed(self) -> str:
raise NotImplementedError
Method name should match the super class method name that you want to override.
New method should take same arguments.
You must add round brackets to decorator.
Example:
from sdbus import (DbusInterfaceCommonAsync, dbus_method_async
dbus_method_async_override)
class ExampleInterface(DbusInterfaceCommonAsync,
interface_name='org.example.test'
):
# Original call
@dbus_method_async('s', 's')
async def upper(self, str_to_up: str) -> str:
return str_to_up.upper()
class ExampleOverride(ExampleInterface):
@dbus_method_async_override()
async def upper(self, str_to_up: str) -> str:
return 'Upper: ' + str_to_up.upper()
You must add round brackets to decorator.
Example:
from sdbus import (DbusInterfaceCommonAsync, dbus_property_async
dbus_property_async_override)
class ExampleInterface(DbusInterfaceCommonAsync,
interface_name='org.example.test'
):
def __init__(self) -> None:
self.s = 'aaaaaaaaa'
# Original property
@dbus_property_async('s')
def str_prop(self) -> str:
return self.s
@str_prop.setter
def str_prop_setter(self, new_s: str) -> None:
self.s = new_s
class ExampleOverride(ExampleInterface):
@dbus_property_async_override()
def str_prop(self) -> str:
return 'Test property' + self.s
# Setter needs to be decorated again to override
@str_prop.setter
def str_prop_setter(self, new_s: str) -> None:
self.s = new_s.upper()
These exceptions are bound to specific D-Bus error names. For example, DbusFailedError is bound to org.freedesktop.DBus.Error.Failed error name.
This means if the remote object sends an error message with this error name the Python will receive this exception.
When raised in a method callback an error message will be sent back to caller.
See list of error exceptions.
If you want to create a new error bound exception you should subclass it from DbusFailedError and provide a unique dbus_error_name attribute in the exception body definition.
Example:
class DbusExampleError(DbusFailedError):
dbus_error_name = 'org.example.Error'
If dbus_error_name is not unique the ValueError will be raised.
Defining an exception will automatically bind incoming error message to this new exception.
Existing exceptions can be manually binded using map_exception_to_dbus_error() function.
All Python built-in exceptions are mapped to D-Bus errors.
The D-Bus error name is created by appending org.python.Error. to the exception name.
For example, AssertionError is bound to org.python.Error.AssertionError name.
The exceptions argument is a tuple of error name and error message.
Exception message contains line number and the error name.
These exceptions will be raise if an error related to ownership of D-Bus names occurs when calling request_default_bus_name_async() or request_default_bus_name().
Recommended to subclass to create a new exception.
Probably should only be raised by bus daemon.
Probably should only be raised by bus daemon.
This is different from DbusNoReplyError as here the connection to bus timeout not the remote object not replying.
Encountered you use D-Bus over TCP or SSH.
Recommended to subclass to create a new exception.
Member names will be translated to python defined names. Invalidated properties will have a value of None.
Takes an iterable of D-Bus interface classes (or a single class) and the signal data. Returns the path of new object, the class of the added object (if it matched one of passed interface classes) and the dictionary of python named properties and their values.
Takes an iterable of D-Bus interface classes (or a single class) and the signal data. Returns the path of removed object andthe class of the added object. (if it matched one of passed interface classes)
In this example we create a simple example server and client.
There are 3 files:
example_interface.py file:
from sdbus import (DbusInterfaceCommonAsync, dbus_method_async,
dbus_property_async, dbus_signal_async)
# This is file only contains interface definition for easy import
# in server and client files
class ExampleInterface(
DbusInterfaceCommonAsync,
interface_name='org.example.interface'
):
@dbus_method_async(
input_signature='s',
result_signature='s',
)
async def upper(self, string: str) -> str:
return string.upper()
@dbus_property_async(
property_signature='s',
)
def hello_world(self) -> str:
return 'Hello, World!'
@dbus_signal_async(
signal_signature='i'
)
def clock(self) -> int:
raise NotImplementedError
example_server.py file:
from asyncio import get_event_loop, sleep
from random import randint
from time import time
from example_interface import ExampleInterface
from sdbus import request_default_bus_name_async
loop = get_event_loop()
export_object = ExampleInterface()
async def clock() -> None:
"""
This coroutine will sleep a random time and emit a signal with current clock
"""
while True:
await sleep(randint(2, 7)) # Sleep a random time
current_time = int(time()) # The interface we defined uses integers
export_object.clock.emit(current_time)
async def startup() -> None:
"""Perform async startup actions"""
# Acquire a known name on the bus
# Clients will use that name to address to this server
await request_default_bus_name_async('org.example.test')
# Export the object to D-Bus
export_object.export_to_dbus('/')
loop.run_until_complete(startup())
task_clock = loop.create_task(clock())
loop.run_forever()
example_client.py file:
from asyncio import get_event_loop
from example_interface import ExampleInterface
# Create a new proxy object
example_object = ExampleInterface.new_proxy('org.example.test', '/')
async def print_clock() -> None:
# Use async for loop to print clock signals we receive
async for x in example_object.clock:
print('Got clock: ', x)
async def call_upper() -> None:
s = 'test string'
s_after = await example_object.upper(s)
print('Initial string: ', s)
print('After call: ', s_after)
async def get_hello_world() -> None:
print('Remote property: ', await example_object.hello_world)
loop = get_event_loop()
# Always binds your tasks to a variable
task_upper = loop.create_task(call_upper())
task_clock = loop.create_task(print_clock())
task_hello_world = loop.create_task(get_hello_world())
loop.run_forever()
Start server before client. python example_server.py
In separated terminal start client. python example_client.py
Use CTRL-C to close client and server.
You can also use ExampleInterface as a local object:
from asyncio import run
from example_interface import ExampleInterface
example_object = ExampleInterface()
async def test() -> None:
print(await example_object.upper('test'))
print(await example_object.hello_world)
run(test())
python-sdbus includes two namespace packages sdbus_async and sdbus_block which are used for proxies.
For example, D-Bus daemon interface (which comes by default) can be found under sdbus_async.dbus_daemon for async binds and sdbus_block.dbus_daemon for blocking binds.
This is the D-Bus daemon interface. Used for querying D-Bus state.
D-Bus interface object path and service name is predetermined. (at 'org.freedesktop.DBus', '/org/freedesktop/DBus')
Get process ID that owns a specified name.
Get process user ID that owns a specified name.
Returns machine id where bus is run. (stored in /etc/machine-id)
Returns unique bus name (i.e. ':1.94') for given service name.
Lists all activatable services names.
List all services and connections currently of the bus.
Return True if someone already owns the name, False if nobody does.
Starts a specified service.
Flags parameter is not used currently and should be omitted or set to 0.
Python type: List[str]
D-Bus type: as
List of D-Bus daemon features.
Features include:
Python type: List[str]
D-Bus type: as
Extra D-Bus daemon interfaces
Python type: str
D-Bus type: s
Signal when current process acquires a bus name.
Python type: str
D-Bus type: s
Signal when current process loses a bus name.
Python type: Tuple[str, str, str]
D-Bus type: sss
Signal when some name on a bus changes owner.
Is a tuple of:
This list contains the known python-sdbus interface collections:
Python-sdbus is able to generate the interfaces code from the D-Bus introspection XML. (either from a file or live object on D-Bus) Currently async interfaces code is generated by default. Blocking interfaces can be generated by passing --block option.
Running code generator requires Jinja2 to be installed.
WARNING:
To run generator on files (such as found under /usr/share/dbus-1/interfaces/ folder) execute the sdbus module with gen-from-file first argument and file paths to introspection XML files:
python -m sdbus gen-from-file /usr/share/dbus-1/interfaces/org.gnome.Shell.Screenshot.xml
The generated interface code will be printed in to stdout. You can use shell redirection > to save it in to file.
Multiple interface files can be passed which generates a file containing multiple interfaces.
To run generator on some service on the D-Bus execute the sdbus module with gen-from-connection first argument, the service connection name as second and one or more object paths:
python -m sdbus gen-from-connection org.freedesktop.systemd1 /org/freedesktop/systemd1
The generated interface code will be printed in to stdout. You can use shell redirection > to save it in to file.
Multiple object paths can be passed which generates a file containing all interfaces encountered in the objects.
Pass --system option to use system bus instead of session bus.
Python-sdbus has an extension for Sphinx autodoc that can document D-Bus interfaces.
To use it include "sdbus.autodoc" extension in your conf.py file.
extensions = ['sdbus.autodoc']
The extension can document interface class bodies. For example, python-sdbus-networkmanager uses it to document the classes.
.. autoclass:: sdbus_async.networkmanager.NetworkManagerDeviceBluetoothInterfaceAsync
:members:
WARNING:
The D-Bus methods should be documented same way as the regular function would. See Sphinx documentation on possible fields
Example docstring for a D-Bus method:
@dbus_method_async('s', method_name='GetConnectionUnixProcessID')
async def get_connection_pid(self, service_name: str) -> int:
"""Get process ID that owns a specified name.
:param service_name: Service name to query.
:return: PID of name owner
:raises DbusNameHasNoOwnerError: Nobody owns that name
"""
raise NotImplementedError
D-Bus properties and signals will be annotated with type taken from the stub function.
@dbus_property_async('as')
def features(self) -> List[str]:
"""List of D-Bus daemon features.
Features include:
* 'AppArmor' - Messages filtered by AppArmor on this bus.
* 'HeaderFiltering' - Messages are filtered if they have incorrect \
header fields.
* 'SELinux' - Messages filtered by SELinux on this bus.
* 'SystemdActivation' - services activated by systemd if their \
.service file specifies a D-Bus name.
"""
raise NotImplementedError
No parameters are supported at the moment for properties and signals.
Python-sdbus provides several utilities to enable unit testing.
Creates an isolated instance of session D-Bus. The D-Bus will be closed and cleaned up after tests are finished.
Requires dbus-daemon executable be installed.
It is also set as a default bus.
Usage example:
from sdbus import DbusInterfaceCommonAsync, dbus_method_async
from sdbus.unittest import IsolatedDbusTestCase
class TestInterface(DbusInterfaceCommonAsync,
interface_name='org.test.test',
):
@dbus_method_async("s", "s")
async def upper(self, string: str) -> str:
"""Uppercase the input"""
return string.upper()
def initialize_object() -> Tuple[TestInterface, TestInterface]:
test_object = TestInterface()
test_object.export_to_dbus('/')
test_object_connection = TestInterface.new_proxy(
"org.example.test", '/')
return test_object, test_object_connection
class TestProxy(IsolatedDbusTestCase):
async def asyncSetUp(self) -> None:
await super().asyncSetUp()
await self.bus.request_name_async("org.example.test", 0)
async def test_method_kwargs(self) -> None:
test_object, test_object_connection = initialize_object()
self.assertEqual(
'TEST',
await test_object_connection.upper('test'),
)
get_default_bus()
request_default_bus_name_async()
set_default_bus()
decode_object_path()
encode_object_path()
sd_bus_open_system()
sd_bus_open_user()
DbusDeprecatedFlag
DbusHiddenFlag
DbusNoReplyFlag
DbusPropertyConstFlag
DbusPropertyEmitsChangeFlag
DbusPropertyEmitsInvalidationFlag
DbusPropertyExplicitFlag
DbusSensitiveFlag
DbusUnprivilegedFlag
DbusInterfaceCommonAsync
dbus_method_async()
dbus_method_async_override()
dbus_property_async()
dbus_property_async_override()
dbus_signal_async()
DbusInterfaceCommon
dbus_method()
dbus_property()
exceptions.DbusAccessDeniedError
exceptions.DbusAccessDeniedError
exceptions.DbusAddressInUseError
exceptions.DbusAuthFailedError
exceptions.DbusBadAddressError
exceptions.DbusDisconnectedError
exceptions.DbusFailedError
exceptions.DbusFileExistsError
exceptions.DbusFileNotFoundError
exceptions.DbusInconsistentMessageError
exceptions.DbusInteractiveAuthorizationRequiredError
exceptions.DbusInvalidArgsError
exceptions.DbusInvalidFileContentError
exceptions.DbusInvalidSignatureError
exceptions.DbusIOError
exceptions.DbusLimitsExceededError
exceptions.DbusMatchRuleInvalidError
exceptions.DbusMatchRuleNotFound
exceptions.DbusNameHasNoOwnerError
exceptions.DbusNoMemoryError
exceptions.DbusNoNetworkError
exceptions.DbusNoReplyError
exceptions.DbusNoServerError
exceptions.DbusNotSupportedError
exceptions.DbusPropertyReadOnlyError
exceptions.DbusServiceUnknownError
exceptions.DbusTimeoutError
exceptions.DbusUnixProcessIdUnknownError
exceptions.DbusUnknownInterfaceError
exceptions.DbusUnknownMethodError
exceptions.DbusUnknownObjectError
exceptions.DbusUnknownPropertyError
exceptions.SdBusBaseError
exceptions.SdBusLibraryError
exceptions.SdBusUnmappedMessageError
exceptions.map_exception_to_dbus_error()
exceptions.SdBusRequestNameError
exceptions.SdBusRequestNameInQueueError
exceptions.SdBusRequestNameExistsError
exceptions.SdBusRequestNameAlreadyOwnerError
igo95862
| April 11, 2024 |