Asyncio API

Classes

class sdbus.DbusInterfaceCommonAsync(interface_name)

Dbus async interface class. Dbus methods and properties should be defined using dbus_property_async(), dbus_signal_async(), and dbus_method_async() decorators.

Note

Don’t forget to call super().__init__() in derived classes init calls as it sets up important attributes.

Parameters:

interface_name (str) – Sets the dbus interface name that will be used for all properties, methods and signals defined in the body of the class.
serving_enabled (bool) – If set to True the interface will not be served on dbus. Mostly used for interfaces that sd-bus already provides such as org.freedesktop.DBus.Peer.

async dbus_ping()

Pings the remote service using dbus.

Useful to test if connection or remote service is alive.

Warning

This method is ignores the particular object path meaning it can NOT be used to test if object exist.

async dbus_machine_id()

Returns the machine UUID of D-Bus the object is connected to.

Returns:: machine UUID
Return type:: str

async dbus_introspect()

Get dbus introspection XML.

It is users responsibility to parse that data.

Returns:: string with introspection XML
Return type:: str

async properties_get_all_dict()

Get all object properties as a dictionary where keys are member names and values are properties values.

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)

Parameters:: on_unknown_member (str) – If an unknown D-Bus property was encountered either raise an "error" (default), "ignore" the property or "reuse" the D-Bus name for the member.
Returns:: dictionary of properties
Return type:: Dict[str, Any]

properties_changed: Tuple[str, Dict[str, Tuple[str, Any]], List[str]]

Signal when one of the objects properties changes.

sdbus.utils.parse_properties_changed() can be used to transform this signal data in to an easier to work with dictionary.

Signal data is:

Interface namestr: Name of the interface where property changed
Changed propertiesDict[str, Tuple[str, Any]]: Dictionary there keys are names of properties changed and values are variants of new value.
Invalidated propertiesList[str]: List of property names changed but no new value had been provided

_proxify(bus, service_name, object_path)

Begin proxying to a remote dbus object.

Parameters:

service_name (str) – Remote object dbus connection name. For example, systemd uses org.freedesktop.systemd1
object_path (str) – Remote object dbus path. Should be a forward slash separated path. Starting object is usually /. Example: /org/freedesktop/systemd/unit/dbus_2eservice
bus (SdBus) – Optional dbus connection object. If not passed the default dbus will be used.

classmethod new_proxy(bus, service_name, object_path)

Create new proxy object and bypass __init__.

Parameters:

service_name (str) – Remote object dbus connection name. For example, systemd uses org.freedesktop.systemd1
object_path (str) – Remote object dbus path. Should be a forward slash separated path. Starting object is usually /. Example: /org/freedesktop/systemd/unit/dbus_2eservice
bus (SdBus) – Optional dbus connection object. If not passed the default dbus will be used.

export_to_dbus(object_path, bus)

Object will appear and become callable on dbus.

Parameters:

object_path (str) – Object path that it will be available at.
bus (SdBus) – Optional dbus connection object. If not passed the default dbus will be used.

class sdbus.DbusObjectManagerInterfaceAsync(interface_name)

This class is almost identical to DbusInterfaceCommonAsync but implements ObjectManager interface.

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

async get_managed_objects()

Get the objects this object manager in managing.

Returns:

Triple nested dictionary that contains all the objects paths with their properties values.

Dict[ObjectPath, Dict[InterfaceName, Dict[PropertyName, PropertyValue]]]

Return type:

Dict[str, Dict[str, Dict[str, Any]]]

interfaces_added: Tuple[str, Dict[str, Dict[str, Any]]]

Signal when a new object is added or and existing object gains a new interface.

sdbus.utils.parse_interfaces_added() can be used to make signal data easier to work with.

Signal data is:

Object pathstr: Path to object that was added or modified.
Object interfaces and propertiesDict[str, Dict[str, Any]]]: Dict[InterfaceName, Dict[PropertyName, PropertyValue]]

interfaces_removed: Tuple[str, List[str]]

Signal when existing object or and interface of existing object is removed.

sdbus.utils.parse_interfaces_removed() can be used to make signal data easier to work with.

Signal data is:

Object pathstr: Path to object that was removed or modified.
Interfaces listList[str]: Interfaces names that were removed.

export_with_manager(object_path, object_to_export, bus)

Export object to D-Bus and emit a signal that it was added.

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.

Parameters:

object_path (str) – Object path that it will be available at.
object_to_export (DbusInterfaceCommonAsync) – Object to export to D-Bus.
bus (SdBus) – Optional dbus connection object. If not passed the default dbus will be used.

Raises:

RuntimeError – ObjectManager was not exported.

remove_managed_object(managed_object)

Emit signal that object was removed.

Releases reference to the object.

Caution

The object will still be accessible over D-Bus until all references to it will be removed.

Parameters:

managed_object (DbusInterfaceCommonAsync) – Object to remove from ObjectManager.

Raises:

RuntimeError – ObjectManager was not exported.
KeyError – Passed object is not managed by ObjectManager.

Decorators

@sdbus.dbus_method_async([input_signature[, result_signature[, flags[, result_args_names[, input_args_names[, method_name]]]]]])

Define a method.

Underlying function must be a coroutine function.

Parameters:

input_signature (str) – dbus input signature. Defaults to “” meaning method takes no arguments. Required if you intend to connect to a remote object.
result_signature (str) – dbus result signature. Defaults to “” meaning method returns empty reply on success. Required if you intend to serve the object.
flags (int) –
modifies behavior. No effect on remote connections. Defaults to 0 meaning no special behavior.

See Flags .
result_args_names (Sequence[str]) –
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.
input_args_names (Sequence[str]) –
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
method_name (str) – Force specific dbus method name instead of being based on Python function name.

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

@sdbus.dbus_property_async(property_signature[, flags[, property_name]])

Declare a dbus property.

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

Properties are supposed to be lightweight to get or set. Make sure property getter or setter does not perform heavy IO or computation as that will block other methods or properties.

Parameters:

property_signature (str) – Property dbus signature. Has to be a single type or container.
flags (int) –
modifies behavior. No effect on remote connections. Defaults to 0 meaning no special behavior.

See Flags .
property_name (str) – Force specific property name instead of constructing it based on Python function name.

Properties have following methods:

@sdbus.setter(set_function)

Defines the setter function. This makes the property read/write instead of read-only.

See example on how to use.

@sdbus.setter_private(set_function)

Defines the private setter function. The setter can be called locally but property will be read-only from D-Bus.

Calling the setter locally will emit properties_changed signal to D-Bus.

async sdbus.get_async()

Get the property value.

The property can also be directly await ed instead of calling this method.

async sdbus.set_async(new_value): Set property value.

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

@sdbus.dbus_signal_async([signal_signature[, signal_args_names[, flags[, signal_name]]]])

Defines a dbus signal.

Underlying function return type hint is used for signal type hints.

Parameters:

signal_signature (str) – signal dbus signature. Defaults to empty signal.
signal_args_names (Sequence[str]) –
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.
flags (int) –
modifies behavior. No effect on remote connections. Defaults to 0 meaning no special behavior.

See Flags .
signal_name (str) – Forces specific signal name instead of being based on Python function name.

Signals have following methods:

sdbus.catch()

Catch D-Bus signals using the async generator for loop: async for x in something.some_signal.catch():

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

sdbus.catch_anywhere(service_name, bus)

Catch signal independent of path. Yields tuple of path of the object that emitted signal and signal data.

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.

Parameters:

service_name (str) – Service name of which signals belong to. Required if called from class. When called from proxy object the service name of the proxy will be used.
bus (str) – Optional dbus connection object. If not passed when called from proxy the bus connected to proxy will be used or when called from class default bus will be used.

sdbus.emit(args): Emit a new signal with args data.

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

@sdbus.dbus_method_async_override

Override the method.

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

@sdbus.dbus_property_async_override

Override property.

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