dasbus Dasbus is a DBus library written in Python 3, based on GLib and inspired by pydbus. The code used to be part of the Anaconda Installer project. It was based on the pydbus library, but we …
You can install PyGObject provided by your system or use PyPI. The system package is usually called python3-gi, python3-gobject or pygobject3. See the instructions for your platform (only for PyGObject, you don't need cairo or GTK).
The library is known to work with Python 3.8, PyGObject 3.34 and GLib 2.63, but these are not the required minimal versions.
Install the package from PyPI. Follow the instructions above to install the required dependencies.
pip3 install dasbus
Or install the RPM package on Fedora 31+.
sudo dnf install python3-dasbus
Look at the complete examples or DBus services of the Anaconda Installer for more inspiration.
Show the current hostname.
from dasbus.connection import SystemMessageBus
bus = SystemMessageBus()
proxy = bus.get_proxy(
"org.freedesktop.hostname1",
"/org/freedesktop/hostname1"
)
print(proxy.Hostname)
Send a notification to the notification server.
from dasbus.connection import SessionMessageBus
bus = SessionMessageBus()
proxy = bus.get_proxy(
"org.freedesktop.Notifications",
"/org/freedesktop/Notifications"
)
id = proxy.Notify(
"", 0, "face-smile", "Hello World!",
"This notification can be ignored.",
[], {}, 0
)
print("The notification {} was sent.".format(id))
Handle a closed notification.
from dasbus.loop import EventLoop
loop = EventLoop()
from dasbus.connection import SessionMessageBus
bus = SessionMessageBus()
proxy = bus.get_proxy(
"org.freedesktop.Notifications",
"/org/freedesktop/Notifications"
)
def callback(id, reason):
print("The notification {} was closed.".format(id))
proxy.NotificationClosed.connect(callback)
loop.run()
Run the service org.example.HelloWorld.
from dasbus.loop import EventLoop
loop = EventLoop()
from dasbus.connection import SessionMessageBus
bus = SessionMessageBus()
class HelloWorld(object):
__dbus_xml__ = """
<node>
<interface name="org.example.HelloWorld">
<method name="Hello">
<arg direction="in" name="name" type="s" />
<arg direction="out" name="return" type="s" />
</method>
</interface>
</node>
"""
def Hello(self, name):
return "Hello {}!".format(name)
bus.publish_object("/org/example/HelloWorld", HelloWorld())
bus.register_service("org.example.HelloWorld")
loop.run()
Use constants to define DBus services and objects.
from dasbus.connection import SystemMessageBus
from dasbus.identifier import DBusServiceIdentifier
NETWORK_MANAGER = DBusServiceIdentifier(
namespace=("org", "freedesktop", "NetworkManager"),
message_bus=SystemMessageBus()
)
proxy = NETWORK_MANAGER.get_proxy()
print(proxy.NetworkingEnabled)
Use exceptions to propagate and handle DBus errors. Create an error mapper and a decorator for mapping Python exception classes to DBus error names. The message bus will use the given error mapper to transform Python exceptions to DBus errors and back.
from dasbus.error import ErrorMapper, DBusError, get_error_decorator
error_mapper = ErrorMapper()
dbus_error = get_error_decorator(error_mapper)
from dasbus.connection import SessionMessageBus
bus = SessionMessageBus(error_mapper=error_mapper)
@dbus_error("org.freedesktop.DBus.Error.InvalidArgs")
class InvalidArgs(DBusError):
pass
Call DBus methods asynchronously.
from dasbus.loop import EventLoop
loop = EventLoop()
def callback(call):
print(call())
proxy = NETWORK_MANAGER.get_proxy()
proxy.GetDevices(callback=callback)
loop.run()
Generate XML specifications from Python classes.
from dasbus.server.interface import dbus_interface
from dasbus.typing import Str
@dbus_interface("org.example.HelloWorld")
class HelloWorld(object):
def Hello(self, name: Str) -> Str:
return "Hello {}!".format(name)
print(HelloWorld.__dbus_xml__)
Represent DBus structures by Python objects.
from dasbus.structure import DBusData
from dasbus.typing import Str, get_variant
class UserData(DBusData):
def __init__(self):
self._name = ""
@property
def name(self) -> Str:
return self._name
@name.setter
def name(self, name):
self._name = name
data = UserData()
data.name = "Alice"
print(UserData.to_structure(data))
print(UserData.from_structure({
"name": get_variant(Str, "Bob")
}))
Create Python objects that can be published on DBus.
from dasbus.server.interface import dbus_interface
from dasbus.server.template import InterfaceTemplate
from dasbus.server.publishable import Publishable
from dasbus.typing import Str
@dbus_interface("org.example.Chat")
class ChatInterface(InterfaceTemplate):
def Send(self, message: Str):
return self.implementation.send()
class Chat(Publishable):
def for_publication(self):
return ChatInterface(self)
def send(self, message):
print(message)
Use DBus containers to publish dynamically created Python objects.
from dasbus.connection import SessionMessageBus
from dasbus.server.container import DBusContainer
container = DBusContainer(
namespace=("org", "example", "Chat"),
message_bus=SessionMessageBus()
)
print(container.to_object_path(Chat()))
The dasbus library used to be based on pydbus, but it was later reimplemented. We have changed the API and the implementation of the library based on our experience with pydbus. However, it should be possible to modify dasbus classes to work the same way as pydbus classes.
The abstract handler of a remote DBus object.
Unsubscribe from DBus signals and disconnect all registered callbacks of the proxy signals.
The client handler of a DBus object.
The low-level DBus client library based on GLib.
Base class for DBus observers.
This class is recommended to use only to watch the availability of a service on DBus. It doesn't provide any support for accessing objects provided by the service.
Usage:
# Create the observer and connect to its signals.
observer = DBusObserver(SystemBus, "org.freedesktop.NetworkManager")
def callback1(observer):
print("Service is available!")
def callback2(observer):
print("Service is unavailable!")
observer.service_available.connect(callback1)
observer.service_unavailable.connect(callback2)
# Connect to the service once it is available.
observer.connect_once_available()
# Disconnect the observer.
observer.disconnect()
The observer is not connected to the service until it emits the service_available signal.
Disconnect from the service if it is connected and stop watching its availability.
Signal emits this class as an argument. You have to call the watch method to activate the signals.
Signal emits this class as an argument. You have to call the watch method to activate the signals.
Exception class for the DBus observers.
The low-level DBus monitoring library based on GLib.
Proxy of a remote DBus property.
It can be used to define instance attributes.
Abstract proxy of a remote DBus object.
Proxy of a remote DBus interface.
Proxy of a remote DBus object.
The container of DBus objects.
A DBus container should be used to dynamically publish Publishable objects within the same namespace. It generates a unique DBus path for each object. It is able to resolve a DBus path into an object and an object into a DBus path.
Example:
# Create a container of tasks.
container = DBusContainer(
namespace=("my", "project"),
basename="Task",
message_bus=DBus
)
# Publish a task.
path = container.to_object_path(MyTask())
# Resolve an object path into a task.
task = container.from_object_path(path)
If no published object is found for the given DBus path, raise DBusContainerError.
All DBus objects from the container should use the same namespace, so the namespace should be set up before any of the DBus objects are published.
If no DBus path is found for the given object, publish the object on the container message bus with a unique DBus path generated from the container namespace.
General exception for DBus container errors.
The abstract handler of a published object.
Handle emitted signals of the object with the _emit_signal method and handle incoming DBus calls with the _handle_call method.
Unregister the object and disconnect all signals.
The low-level DBus server library based on GLib.
Supported items:
There can be more supported items in the future.
The handler of an object published on DBus.
The decorator allows the server object handler to propagate additional information about the DBus call into the decorated method.
Use a dictionary of keyword arguments:
@accepts_additional_arguments
def Method(x: Int, y: Str, **info):
pass
Or use keyword only parameters:
@accepts_additional_arguments
def Method(x: Int, y: Str, *, call_info):
pass
At this moment, the library provides only the call_info argument generated by GLibServer.get_call_info, but the additional arguments can be customized in the _get_additional_arguments method of the server object handler.
A new DBus class can be defined as:
@dbus_class
class Class(Interface):
...
DBus class can implement DBus interfaces, but it cannot define a new interface.
The DBus XML specification will be generated from implemented interfaces (inherited) and it will be accessible as:
Class.__dbus_xml__
A new DBus interface can be defined as:
@dbus_interface
class Interface():
...
The interface will be generated from the given class cls with a name interface_name and added to the DBus XML specification of the class.
The XML specification is accessible as: .. code-block:: python
DBus signal.
Can be used as:
Signal = dbus_signal()
Or as a method decorator:
@dbus_signal
def Signal(x: Int, y: Double):
pass
Signal is defined by the type hints of a decorated method. This method is accessible as: signal.definition
If the signal is not defined by a method, it is expected to have no arguments and signal.definition is equal to None.
Exception for DBus properties.
Standard DBus interface org.freedesktop.DBus.Properties.
DBus objects don't have to inherit this class, because the DBus library provides support for this interface by default. This class only extends this support.
Report the changed property:
self.report_changed_property('X')
Emit all changes when the method is done:
@emits_properties_changed
def SetX(x: Int):
self.set_x(x)
Can be used as:
Signal = dbus_signal()
Or as a method decorator:
@dbus_signal
def Signal(x: Int, y: Double):
pass
Signal is defined by the type hints of a decorated method. This method is accessible as: signal.definition
If the signal is not defined by a method, it is expected to have no arguments and signal.definition is equal to None.
The decorated method has to be a member of a class that inherits PropertiesInterface.
Abstract class for Python objects that can be published on DBus.
Example:
# Define a publishable class.
class MyObject(Publishable):
def for_publication(self):
return MyDBusInterface(self)
# Create a publishable object.
my_object = MyObject()
# Publish the object on DBus.
DBus.publish_object("/org/project/x", my_object.for_publication())
Basic template for a DBus interface.
This template uses a software design pattern called proxy.
This class provides a recommended way how to define DBus interfaces and create publishable DBus objects. The class that defines a DBus interface should inherit this class and be decorated with @dbus_class or @dbus_interface decorator. The implementation of this interface will be provided by a separate object called implementation. Therefore the methods of this class should call the methods of the implementation, the signals should be connected to the signals of the implementation and the getters and setters of properties should access the properties of the implementation.
@dbus_interface("org.myproject.X")
class InterfaceX(BasicInterfaceTemplate):
def DoSomething(self) -> Str:
return self.implementation.do_something()
class X(object):
def do_something(self):
return "Done!"
x = X()
i = InterfaceX(x)
DBus.publish_object("/org/myproject/X", i)
You should connect the emit methods of the interface signals to the signals of the implementation. Every time the implementation emits a signal, this interface reemits the signal on DBus.
Template for a DBus interface.
The interface provides the support for the standard interface org.freedesktop.DBus.Properties.
Usage:
def connect_signals(self):
super().connect_signals()
self.implementation.module_properties_changed.connect(
self.flush_changes
)
self.watch_property("X", self.implementation.x_changed)
@property
def X(self, x) -> Int:
return self.implementation.x
@emits_properties_changed
def SetX(self, x: Int):
self.implementation.set_x(x)
Report a change when the property is changed.
Representation of a connection for the specified address.
The low-level DBus connection library based on GLib.
Representation of a message bus based on D-Bus.
If the proxy factory is not specified, we will use a default one. If the interface name is set, we will choose InterfaceProxy, otherwise ObjectProxy.
If the interface name is set, we will add it to the additional arguments for the proxy factory.
Representation of a session bus connection.
Representation of a system bus connection.
Abstract rule for mapping a Python exception to a DBus error.
param error_name: a name of the DBus error :return: a type of the Python error
A default DBus error.
Default rule for mapping a Python exception to a DBus error.
Class for mapping Python exceptions to DBus errors.
The new rule will have a higher priority than the rules already contained in the error mapper.
Try to find a matching rule in the error mapper. If a rule matches the given exception type, use the rule to get the name of the DBus error.
The rules in the error mapper are processed in the reversed order to respect the priority of the rules.
Try to find a matching rule in the error mapper. If a rule matches the given name of a DBus error, use the rule to get the type of a Python exception.
The rules in the error mapper are processed in the reversed order to respect the priority of the rules.
Reset the error rules to the initial state. All rules will be replaced with the default ones.
Rule for mapping a Python exception to a DBus error.
Create a function for decorating Python exception classes. The decorator will add a new rule to the given error mapper that will map the class to the specified error name.
Definition of the decorator:
decorator(error_name, namespace=())
The decorator accepts a name of the DBus error and optionally a namespace of the DBus name. The namespace will be used as a prefix of the DBus name.
Usage:
# Create an error mapper.
error_mapper = ErrorMapper()
# Create a decorator for DBus errors and use it to map
# the class ExampleError to the name my.example.Error.
dbus_error = create_error_decorator(error_mapper)
@dbus_error("my.example.Error")
class ExampleError(DBusError):
pass
Identifier of a DBus interface.
Identifier of a DBus object.
Identifier of a DBus service.
If no object path is specified, we will use the object path of this DBus service.
If no interface name is specified, we will use none and create a proxy from all interfaces of the DBus object.
The abstract representation of the event loop.
It is necessary to run the event loop to handle emitted DBus signals or incoming DBus calls (in the DBus service).
Example:
# Create the event loop. loop = EventLoop() # Start the event loop. loop.run() # Run loop.quit() to stop.
The representation of the event loop.
Default representation of a signal.
If no callback is specified, then all functions will be unregistered from the signal.
If the specified callback isn't registered, do nothing.
DBus XML specification.
Exception for the DBus specification errors.
Class for parsing DBus XML specification.
Object representation of data in a DBus structure.
Classes derived from this class should represent specific types of DBus structures. They will support a conversion from a DBus structure of this type to a Python object and back.
General exception for DBus structure errors.
Set the argument 'skip' to skip attributes with sensitive data.
Set the argument 'add' to add other values to the string representation. The attributes in the string representation will be sorted alphabetically.
new_array(child_type:GLib.VariantType=None, children:list=None) -> GLib.Variant new_boolean(value:bool) -> GLib.Variant new_byte(value:int) -> GLib.Variant new_bytestring(string:list) -> GLib.Variant new_bytestring_array(strv:list) -> GLib.Variant new_dict_entry(key:GLib.Variant, value:GLib.Variant) -> GLib.Variant new_double(value:float) -> GLib.Variant new_fixed_array(element_type:GLib.VariantType, elements=None, n_elements:int, element_size:int) -> GLib.Variant new_from_bytes(type:GLib.VariantType, bytes:GLib.Bytes, trusted:bool) -> GLib.Variant new_from_data(type:GLib.VariantType, data:list, trusted:bool, notify:GLib.DestroyNotify, user_data=None) -> GLib.Variant new_handle(value:int) -> GLib.Variant new_int16(value:int) -> GLib.Variant new_int32(value:int) -> GLib.Variant new_int64(value:int) -> GLib.Variant new_maybe(child_type:GLib.VariantType=None, child:GLib.Variant=None) -> GLib.Variant new_object_path(object_path:str) -> GLib.Variant new_objv(strv:list) -> GLib.Variant new_signature(signature:str) -> GLib.Variant new_string(string:str) -> GLib.Variant new_strv(strv:list) -> GLib.Variant new_tuple(children:list) -> GLib.Variant new_uint16(value:int) -> GLib.Variant new_uint32(value:int) -> GLib.Variant new_uint64(value:int) -> GLib.Variant new_variant(value:GLib.Variant) -> GLib.Variant
If the signature is not a tuple, it returns one element with the entire signature. If the signature is an empty tuple, the result is [].
This is useful for e. g. iterating over method parameters which are passed as a single Variant.
new(type_string:str) -> GLib.VariantType new_array(element:GLib.VariantType) -> GLib.VariantType new_dict_entry(key:GLib.VariantType, value:GLib.VariantType) -> GLib.VariantType new_maybe(element:GLib.VariantType) -> GLib.VariantType new_tuple(items:list) -> GLib.VariantType
This function is useful for testing, when the DBus library doesn't decompose arguments and return values of DBus calls.
For example, Str and Int are arguments of the type hint Tuple(Str, Int).
The type of a variant is specified with a type hint.
Example:
v1 = get_variant(Bool, True) v2 = get_variant(List[Int], [1,2,3])
For example, List is a base of the type hint List[Int] and Int is a base of the type hint Int. A class is a base of itself and of every subclass of this class.
Unlike the unpack method of the Variant class, this function doesn't recursively unpacks all variants in the data structure. It will unpack only the topmost variant.
The implementation is inspired by the unpack method.
Class for generating XML.
Python 3.8 changed the order of the attributes and introduced the function canonicalize that should be used to normalize XML.
Class for parsing XML.
Vendula Poncova
2021, Vendula Poncova