SImple (but specific) listener and sender Python 3 DBus example

Question

I want to make a program that has two parts. A listener (a server, if you will) and a sender (the client). I did some research and learned that this is done via a method programmers call IPC (inter process communication); I'm sure you know what it means, I am just expanding the acronym so that you know that I don't think it means Internet Pet Cannibals (or some other non related unpleasant thing).

I read that a good way to achieve this is to use dbus. So I did some research on dbus, and now I'm just confused. Apparently there are a lot of things you can do with dbus, like send notifications to Gnome Shell or talk with the Network Manager. I don't want to do those things! I just want to make two simple programs that talk to each other. To add to that, some tutorials and documentation show examples with python 2, some use 3, some import dbus and some import Gio! A lot of the information I have found is over my head which also impairs my efforts.

Would someone be so kind as to show me a simple, elegant example on how to achieve making a program that essentially does this:

$ ./server
Server is not running yet. Putting on listening ears.
$ ./client Hi
server: a client said "Hi"
$ ./server
Server is already running.
$ ./server stop
Server exiting...
$ ./client Do a barrel roll
client: No one can hear me!!

This is how a simple session would go (using a bash shell of course). I would want to use Python 3 and whatever dbus bindings are most appropriate as of now (I am guessing that would be gi.repository). To clarify, this would be for Linux.

Answer 1

There's not a lot of documentation for dbus in python3, but I managed to figure it out so I'll document it here: The major difference from all the python2 examples is replacing import gobject with import gi.repository.GLib.

You can find more examples (which use more features than I needed) in the dbus-python examples directory.

I didn't implement self-backgrounding in the server because that style of daemon has gone out of style recently.

common.py:

# well-known name for our program
ECHO_BUS_NAME = 'com.stackoverflow.question_21793826.EchoService'

# interfaces implemented by some objects in our program
ECHO_INTERFACE = 'com.stackoverflow.question_21793826.EchoInterface'
QUIT_INTERFACE = 'com.stackoverflow.question_21793826.QuitInterface'

# paths to some objects in our program
ECHO_OBJECT_PATH = '/EchoServerObject'

server.py:

#!/usr/bin/env python3

# standard includes
import sys

# dbus includes
import gi.repository.GLib
import dbus
import dbus.service
import dbus.mainloop.glib

# project includes
import common


class EchoServerObject(dbus.service.Object):

    # TODO it would be nice to make a better decorator using annotations:
    #   def foo(self, a: 's', b: 's') -> '': pass
    # but the existing dbus decorator does its own reflection which
    # fails if there are any annotations (or keyword-only arguments)
    @dbus.service.method(common.ECHO_INTERFACE,
            in_signature='s', out_signature='')
    def echo(self, message):
        message = str(message) # get rid of subclass for repr
        print('server: a client said %r' % message)

    @dbus.service.method(common.QUIT_INTERFACE,
            in_signature='', out_signature='')
    def quit(self):
        # this should be a separate object, but I'm
        # showing how one object can have multiple interfaces
        self.mainloop.quit()

def stop():
    bus = dbus.SessionBus()

    proxy = bus.get_object(common.ECHO_BUS_NAME, common.ECHO_OBJECT_PATH)
    iface = dbus.Interface(proxy, common.QUIT_INTERFACE)

    iface.quit()

def server():
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    bus = dbus.SessionBus()
    try:
        name = dbus.service.BusName(common.ECHO_BUS_NAME, bus, do_not_queue=True)
    except dbus.NameExistsException:
        sys.exit('Server is already running.')
    else:
        print('Server is not running yet. Putting on listening ears.')
    echo = EchoServerObject(bus, common.ECHO_OBJECT_PATH)

    mainloop = gi.repository.GLib.MainLoop()
    echo.mainloop = mainloop
    mainloop.run()

def main(exe, args):
    if args == ['stop']:
        stop()
    elif not args:
        server()
    else:
        sys.exit('Usage: %s [stop]' % exe)

if __name__ == '__main__':
    main(sys.argv[0], sys.argv[1:])

client.py:

#!/usr/bin/env python3

# standard includes
import sys

# dbus includes
import dbus

# project includes
import common


def client(mes):
    bus = dbus.SessionBus()

    try:
        proxy = bus.get_object(common.ECHO_BUS_NAME, common.ECHO_OBJECT_PATH)
    except dbus.DBusException as e:
        # There are actually two exceptions thrown:
        # 1: org.freedesktop.DBus.Error.NameHasNoOwner
        #   (when the name is not registered by any running process)
        # 2: org.freedesktop.DBus.Error.ServiceUnknown
        #   (during auto-activation since there is no .service file)
        # TODO figure out how to suppress the activation attempt
        # also, there *has* to be a better way of managing exceptions
        if e._dbus_error_name != 'org.freedesktop.DBus.Error.ServiceUnknown':
            raise
        if e.__context__._dbus_error_name != 'org.freedesktop.DBus.Error.NameHasNoOwner':
            raise
        print('client: No one can hear me!!')
    else:
        iface = dbus.Interface(proxy, common.ECHO_INTERFACE)
        iface.echo(mes)

def main(exe, args):
    if args:
        client(' '.join(args))
    else:
        sys.exit('Usage: %s message...' % exe)

if __name__ == '__main__':
    main(sys.argv[0], sys.argv[1:])