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.