The D-Bus library is a messaging library used by various desktop environments (GNOME, KDE, etc) for interprocess communication.
There are multiple Python bindings for DBus:
GDbus and QtDbus are wrappers over the C/C++ APIs of GLib and Qt
pydbus is a modern, pythonic API with the goal of being as invisible a layer between the exported API and its user as possible
dasbus is a Python3-only alternative to pydbus with additional features and better flexibility
dbus-python is a legacy API, built with a deprecated dbus-glib library, and involving a lot of type-guessing (despite "explicit is better than implicit" and "resist the temptation to guess").
txdbus is a native Python implementation of the D-Bus protocol for the Twisted networking framework.
dbus-next is a native Python library for D-Bus with support for multiple IO backends suitable for GUI develeopment and cross platform projects.
python-sdbus modern D-Bus library supporting both asyncio and blocking calls, unified client-server classes and type hints.
See also: DBusBindings on Freedesktop wiki.
The dbus-viewer and qdbusviewer programs let you browse through the services and interfaces available on your system.
pydbus
For more information see pydbus's Readme.
Introspection
from pydbus import SessionBus bus = SessionBus() # Create an object that will proxy for a particular remote object. remote_object = bus.get( "org.freedesktop.DBus", # Bus name "/org/freedesktop/DBus" # Object path ) # Introspection returns an XML document containing information # about the methods supported by an interface. print("Introspection data:\n") print(remote_object.Introspect())Output:
Introspection data: <!DOCTYPE node PUBLIC "-//freedesktop//DTD D-BUS Object Introspection 1.0//EN" "http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd"> <node> <interface name="org.freedesktop.DBus.Introspectable"> <method name="Introspect"> <arg name="data" direction="out" type="s"/> </method> </interface> <interface name="org.freedesktop.DBus"> <method name="RequestName"> <arg direction="in" type="s"/> <arg direction="in" type="u"/> <arg direction="out" type="u"/> </method> ...
Calling an interface method
After executing the introspection example:
print(remote_object.ListNames())
Output:
['org.freedesktop.DBus', 'org.freedesktop.Notifications', 'org.freedesktop.PowerManagement', ':1.8', ':1.9', 'org.kde.kaccess', 'org.kde.kded', 'org.kde.StatusNotifierItem-655-1', 'org.freedesktop.systemd1', 'org.ktorrent.ktorrent', 'org.kde.StatusNotifierItem-656-1', 'org.kde.konversation', 'org.pulseaudio.Server', 'org.kde.KScreen', 'org.kde.krunner', 'org.kde.konsole', ':1.40', 'org.a11y.Bus', ':1.41', ':1.42', ':1.20', ':1.43', 'org.kde.klauncher5', ':1.21', ':1.23', 'org.kde.dolphin-3012', 'org.freedesktop.PowerManagement.Inhibit', 'org.kde.Solid.PowerManagement', ':1.24', ':1.25', ':1.49', 'org.kde.kmix', 'org.kde.screensaver', 'org.kde.KWin', 'org.bluez.obex', ':1.29', 'ca.desrt.dconf', 'org.kde.kgpg', 'org.freedesktop.ScreenSaver', 'org.kde.plasmashell', 'org.kde.plasmanetworkmanagement', 'org.kde.StatusNotifierItem-666-1', 'org.kde.kglobalaccel', 'org.freedesktop.FileManager1', 'org.kde.kwalletd5', 'org.PulseAudio1', 'org.kde.polkit-kde-authentication-agent-1', ':1.93', 'org.kde.kded5', 'org.kde.ActivityManager', 'org.kde.keyboard', 'org.kde.kate-3030', ':1.31', 'org.kde.kuiserver', ':1.32', ':1.55', ':1.33', ':1.11', 'org.kde.kwin.Screenshot', ':1.56', ':1.34', 'org.kde.StatusNotifierWatcher', 'org.kde.JobViewServer', ':1.35', ':1.0', ':1.13', ':1.58', 'org.kde.StatusNotifierHost-616', ':1.14', ':1.59', ':1.15', ':1.38', ':1.16', 'org.kde.ksmserver', ':1.39', ':1.17', ':1.5', 'org.kde.Solid.PowerManagement.PolicyAgent', ':1.18', 'org.kde.klauncher', ':1.6', ':1.19']
The following example makes your system hibernate:
# Get the power management object power = bus.get('org.gnome.PowerManager', '/org/gnome/PowerManager') # Hibernate the system if power.CanHibernate(): power.Hibernate()
dasbus
For more information see dasbus's documentation.
Introspection of a remote object
Introspection returns an XML string containing information about interfaces, methods, properties and signals of the remote object.
from dasbus.connection import SessionMessageBus bus = SessionMessageBus() # Create an object that will be a proxy for a particular remote object. remote_object = bus.get_proxy( "org.freedesktop.DBus", # The bus name "/org/freedesktop/DBus" # The object path ) # Call the Introspect method of the remote object. print(remote_object.Introspect())
Accessing a remote property
The following example prints the current hostname.
from dasbus.connection import SystemMessageBus bus = SystemMessageBus() proxy = bus.get_proxy( "org.freedesktop.hostname1", "/org/freedesktop/hostname1" ) print(proxy.Hostname)
Calling a remote method
The following example sends 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", "My notification", "Hello World!", [], {}, 0 ) print("The notification {} was sent.".format(id))
dbus-next
For more information see dbus-next's documentation.
The client interface
To use a service on the bus, the library constructs a proxy object you can use to call methods, get and set properties, and listen to signals.
This example connects to a media player and controls it with the MPRIS DBus interface using python's asyncio backend.
1 from dbus_next.aio import MessageBus 2 3 import asyncio 4 5 loop = asyncio.get_event_loop() 6 7 8 async def main(): 9 bus = await MessageBus().connect() 10 # the introspection xml would normally be included in your project, but 11 # this is convenient for development 12 introspection = await bus.introspect('org.mpris.MediaPlayer2.vlc', '/org/mpris/MediaPlayer2') 13 14 obj = bus.get_proxy_object('org.mpris.MediaPlayer2.vlc', '/org/mpris/MediaPlayer2', introspection) 15 player = obj.get_interface('org.mpris.MediaPlayer2.Player') 16 properties = obj.get_interface('org.freedesktop.DBus.Properties') 17 18 # call methods on the interface (this causes the media player to play) 19 await player.call_play() 20 21 volume = await player.get_volume() 22 print(f'current volume: {volume}, setting to 0.5') 23 24 await player.set_volume(0.5) 25 26 # listen to signals 27 def on_properties_changed(interface_name, changed_properties, invalidated_properties): 28 for changed, variant in changed_properties.items(): 29 print(f'property changed: {changed} - {variant.value}') 30 31 properties.on_properties_changed(on_properties_changed) 32 33 await loop.create_future() 34 35 loop.run_until_complete(main())
The service interface
To define a service on the bus, use the ServiceInterface class and decorate class methods to specify DBus methods, properties, and signals with their type signatures.
1 from dbus_next.service import ServiceInterface, method, dbus_property, signal, Variant 2 from dbus_next.aio MessageBus 3 4 import asyncio 5 6 class ExampleInterface(ServiceInterface): 7 def __init__(self, name): 8 super().__init__(name) 9 self._string_prop = 'kevin' 10 11 @method() 12 def Echo(self, what: 's') -> 's': 13 return what 14 15 @method() 16 def GetVariantDict() -> 'a{sv}': 17 return { 18 'foo': Variant('s', 'bar'), 19 'bat': Variant('x', -55), 20 'a_list': Variant('as', ['hello', 'world']) 21 } 22 23 @dbus_property() 24 def string_prop(self) -> 's': 25 return self._string_prop 26 27 @string_prop.setter 28 def string_prop_setter(self, val: 's'): 29 self._string_prop = val 30 31 @signal() 32 def signal_simple(self) -> 's': 33 return 'hello' 34 35 async def main(): 36 bus = await MessageBus().connect() 37 interface = ExampleInterface('test.interface') 38 bus.export('/test/path', interface) 39 # now that we are ready to handle requests, we can request name from D-Bus 40 await bus.request_name('test.name') 41 # wait indefinitely 42 await asyncio.get_event_loop().create_future() 43 44 asyncio.get_event_loop().run_until_complete(main())
python-sdbus
See python-sdbus documentation for API reference and quickstart guides.
Python-sdbus has a unified client and server classes where one class can be used both as a client proxy and as a server.
Interface class definition
1 from sdbus import (DbusInterfaceCommonAsync, dbus_method_async, 2 dbus_property_async, dbus_signal_async) 3 4 # This is file only contains interface definition for easy import 5 # in server and client files 6 7 class ExampleInterface( 8 DbusInterfaceCommonAsync, 9 interface_name='org.example.interface' 10 ): 11 @dbus_method_async( 12 input_signature='s', 13 result_signature='s', 14 ) 15 async def upper(self, string: str) -> str: 16 return string.upper() 17 18 @dbus_property_async( 19 property_signature='s', 20 ) 21 def hello_world(self) -> str: 22 return 'Hello, World!' 23 24 @dbus_signal_async( 25 signal_signature='i' 26 ) 27 def clock(self) -> int: 28 raise NotImplementedError
Server
1 from asyncio import new_event_loop, sleep 2 from random import randint 3 from time import time 4 5 from example_interface import ExampleInterface 6 7 from sdbus import request_default_bus_name_async 8 9 loop = new_event_loop() 10 11 export_object = ExampleInterface() 12 13 14 async def clock() -> None: 15 """ 16 This coroutine will sleep a random time and emit 17 a signal with current clock 18 """ 19 while True: 20 await sleep(randint(2, 7)) # Sleep a random time 21 current_time = int(time()) # The interface we defined uses integers 22 export_object.clock.emit(current_time) 23 24 25 async def startup() -> None: 26 """Perform async startup actions""" 27 # Acquire a known name on the bus 28 # Clients will use that name to address this server 29 await request_default_bus_name_async('org.example.test') 30 # Export the object to D-Bus 31 export_object.export_to_dbus('/') 32 33 34 loop.run_until_complete(startup()) 35 task_clock = loop.create_task(clock()) 36 loop.run_forever()
Client
1 from asyncio import new_event_loop 2 3 from example_interface import ExampleInterface 4 5 # Create a new proxied object 6 example_object = ExampleInterface.new_proxy('org.example.test', '/') 7 8 9 async def print_clock() -> None: 10 # Use async for loop to print clock signals we receive 11 async for x in example_object.clock: 12 print('Got clock: ', x) 13 14 15 async def call_upper() -> None: 16 s = 'test string' 17 s_after = await example_object.upper(s) 18 19 print('Initial string: ', s) 20 print('After call: ', s_after) 21 22 23 async def get_hello_world() -> None: 24 print('Remote property: ', await example_object.hello_world) 25 26 loop = new_event_loop() 27 28 # Always bind your tasks to a variable 29 task_upper = loop.create_task(call_upper()) 30 task_clock = loop.create_task(print_clock()) 31 task_hello_world = loop.create_task(get_hello_world()) 32 33 loop.run_forever()