Connecting to DBus and sending messages¶
So far, Jeepney can be used with three different I/O systems:
For each of these, there is a module in jeepney.integrate
which exposes
a function called connect_and_authenticate
. This establishes a DBus
connection and returns an object you can use to send and receive messages.
Exactly what it returns may vary, though.
Here’s an example of sending a desktop notification, using blocking I/O:
from jeepney import DBusAddress, new_method_call
from jeepney.integrate.blocking import connect_and_authenticate
notifications = DBusAddress('/org/freedesktop/Notifications',
bus_name='org.freedesktop.Notifications',
interface='org.freedesktop.Notifications')
connection = connect_and_authenticate(bus='SESSION')
# Construct a new D-Bus message. new_method_call takes the address, the
# method name, the signature string, and a tuple of arguments.
msg = new_method_call(notifications, 'Notify', 'susssasa{sv}i',
('jeepney_test', # App name
0, # Not replacing any previous notification
'', # Icon
'Hello, world!', # Summary
'This is an example notification from Jeepney',
[], {}, # Actions, hints
-1, # expire_timeout (-1 = default)
))
# Send the message and wait for the reply
reply = connection.send_and_get_reply(msg)
print('Notification ID:', reply[0])
connection.close()
And here is the same thing using asyncio:
import asyncio
from jeepney import DBusAddress, new_method_call
from jeepney.integrate.asyncio import connect_and_authenticate
notifications = DBusAddress('/org/freedesktop/Notifications',
bus_name='org.freedesktop.Notifications',
interface='org.freedesktop.Notifications')
async def send_notification():
(transport, protocol) = await connect_and_authenticate(bus='SESSION')
msg = new_method_call(notifications, 'Notify', 'susssasa{sv}i',
('jeepney_test', # App name
0, # Not replacing any previous notification
'', # Icon
'Hello, world!', # Summary
'This is an example notification from Jeepney',
[], {}, # Actions, hints
-1, # expire_timeout (-1 = default)
))
# Send the message and await the reply
reply = await protocol.send_message(msg)
print('Notification ID:', reply[0])
loop = asyncio.get_event_loop()
loop.run_until_complete(send_notification())
Message generators and proxies¶
If you’re calling a number of different methods, you can make a message generator class containing their definitions. Jeepney includes a tool to generate these classes automatically—see Generating D-Bus wrappers.
Message generators define how to construct messages. Proxies are wrappers around message generators which send a message and get the reply back.
Let’s rewrite the example above to use a message generator and a proxy:
import asyncio
from jeepney import MessageGenerator, new_method_call
from jeepney.integrate.asyncio import connect_and_authenticate, Proxy
# ---- Message generator, created by jeepney.bindgen ----
class Notifications(MessageGenerator):
interface = 'org.freedesktop.Notifications'
def __init__(self, object_path='/org/freedesktop/Notifications',
bus_name='org.freedesktop.Notifications'):
super().__init__(object_path=object_path, bus_name=bus_name)
def Notify(self, arg_0, arg_1, arg_2, arg_3, arg_4, arg_5, arg_6, arg_7):
return new_method_call(self, 'Notify', 'susssasa{sv}i',
(arg_0, arg_1, arg_2, arg_3, arg_4, arg_5, arg_6, arg_7))
def CloseNotification(self, arg_0):
return new_method_call(self, 'CloseNotification', 'u',
(arg_0,))
def GetCapabilities(self):
return new_method_call(self, 'GetCapabilities')
def GetServerInformation(self):
return new_method_call(self, 'GetServerInformation')
# ---- End auto generated code ----
async def send_notification():
(transport, protocol) = await connect_and_authenticate(bus='SESSION')
proxy = Proxy(Notifications(), protocol)
resp = await proxy.Notify('jeepney_test', # App name
0, # Not replacing any previous notification
'', # Icon
'Hello, world!', # Summary
'This is an example notification from Jeepney',
[], {}, # Actions, hints
-1, # expire_timeout (-1 = default)
)
print('Notification ID:', resp[0])
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(send_notification())
This is more code for the simple use case here, but in a larger application collecting the message definitions together like this could make it clearer.