Examples

Look at the complete examples or DBus services of the Anaconda Installer for more inspiration.

Basic usage

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

Asynchronously fetch a list of network devices.

from dasbus.loop import EventLoop
loop = EventLoop()

from dasbus.connection import SystemMessageBus
bus = SystemMessageBus()

proxy = bus.get_proxy(
    "org.freedesktop.NetworkManager",
    "/org/freedesktop/NetworkManager"
)

def callback(call):
    print(call())

proxy.GetDevices(callback=callback)
loop.run()

Define the org.example.HelloWorld service.

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)

Define the org.example.HelloWorld service with an automatically generated XML specification.

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

Publish the org.example.HelloWorld service on the session message bus.

from dasbus.connection import SessionMessageBus
bus = SessionMessageBus()
bus.publish_object("/org/example/HelloWorld", HelloWorld())
bus.register_service("org.example.HelloWorld")

from dasbus.loop import EventLoop
loop = EventLoop()
loop.run()

Support for Unix file descriptors

The support for Unix file descriptors is disabled by default. It needs to be explicitly enabled when you create a DBus proxy or publish a DBus object that could send or receive Unix file descriptors.

Warning

This functionality is supported only on UNIX.

Send and receive Unix file descriptors with a DBus proxy.

import os
from dasbus.connection import SystemMessageBus
from dasbus.unix import GLibClientUnix
bus = SystemMessageBus()

proxy = bus.get_proxy(
    "org.freedesktop.login1",
    "/org/freedesktop/login1",
    client=GLibClientUnix
)

fd = proxy.Inhibit(
    "sleep", "my-example", "Running an example", "block"
)

proxy.ListInhibitors()
os.close(fd)

Allow to send and receive Unix file descriptors within the /org/example/HelloWorld DBus object.

from dasbus.unix import GLibServerUnix
bus.publish_object(
    "/org/example/HelloWorld",
    HelloWorld(),
    server=GLibServerUnix
)

Management of DBus names and paths

Use constants to define DBus services and objects.

from dasbus.connection import SystemMessageBus
from dasbus.identifier import DBusServiceIdentifier, DBusObjectIdentifier

NETWORK_MANAGER_NAMESPACE = (
    "org", "freedesktop", "NetworkManager"
)

NETWORK_MANAGER = DBusServiceIdentifier(
    namespace=NETWORK_MANAGER_NAMESPACE,
    message_bus=SystemMessageBus()
)

NETWORK_MANAGER_SETTINGS = DBusObjectIdentifier(
    namespace=NETWORK_MANAGER_NAMESPACE,
    basename="Settings"
)

Create a proxy of the org.freedesktop.NetworkManager service.

proxy = NETWORK_MANAGER.get_proxy()
print(proxy.NetworkingEnabled)

Create a proxy of the /org/freedesktop/NetworkManager/Settings object.

proxy = NETWORK_MANAGER.get_proxy(NETWORK_MANAGER_SETTINGS)
print(proxy.Hostname)

See a complete example.

Error handling

Use exceptions to propagate and handle DBus errors. Create an error mapper and a decorator for mapping Python exception classes to DBus error names.

from dasbus.error import ErrorMapper, DBusError, get_error_decorator
error_mapper = ErrorMapper()
dbus_error = get_error_decorator(error_mapper)

Use the decorator to register Python exceptions that represent DBus errors. These exceptions can be raised by DBus services and caught by DBus clients in the try-except block.

@dbus_error("org.freedesktop.DBus.Error.InvalidArgs")
class InvalidArgs(DBusError):
    pass

The message bus will use the specified error mapper to automatically transform Python exceptions to DBus errors and back.

from dasbus.connection import SessionMessageBus
bus = SessionMessageBus(error_mapper=error_mapper)

See a complete example.

Timeout for a DBus call

Call DBus methods with a timeout (specified in milliseconds).

proxy = NETWORK_MANAGER.get_proxy()

try:
    proxy.CheckConnectivity(timeout=3)
except TimeoutError:
    print("The call timed out!")

Support for DBus structures

Represent DBus structures by Python objects. A DBus structure is a dictionary of attributes that maps attribute names to variants with attribute values. Use Python objects to define such structures. They can be easily converted to a dictionary, send via DBus and converted back to an object.

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")
}))

See a complete example.

Management of dynamic DBus objects

Create Python objects that can be automatically published on DBus. These objects are usually managed by DBus containers and published on demand.

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 automatically publish dynamically created Python objects. A DBus container converts publishable Python objects into DBus paths and back. It generates unique DBus paths in the specified namespace and assigns them to objects. Each object is published when its DBus path is requested for the first time.

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

See a complete example.