a threaded processing queue in PyGTK

I'm currently writing a PyGTK client that needs to make network requests using a library that doesn't integrate with the GLib mainloop (python-suds), so I found myself wanting to be able to make network requests without blocking the mainloop, and getting callbacks in my main thread when operations were done. The pattern to use is clearly having a dedicated network thread. In C I might have used GAsyncQueue, however I've found myself quite liking queue.Queue.

The following is a fairly generic class for queuing asynchronous requests. Calling the add_request() method from the main thread queues a function to be run in the worker thread. If the callback or error keywords are provided, these will then be called from the GLib mainloop in the main thread (queued via g_idle_add).

from threading import Thread
from Queue import Queue

import gobject

class ThreadQueue(object):
    def __init__(self):
        self.q = Queue()

        t = Thread(target=self._thread_worker)
        t.setDaemon(True)
        t.start()

    def add_request(self, func, *args, **kwargs):
        """Add a request to the queue. Pass callback= and/or error= as
           keyword arguments to receive return from functions or exceptions.
        """

        self.q.put((func, args, kwargs))

    def _thread_worker(self):
        while True:
            request = self.q.get()
            self.do_request(request)
            self.q.task_done()

    def do_request(self, (func, args, kwargs)):
        if 'callback' in kwargs:
            callback = kwargs['callback']
            del kwargs['callback']
        else:
            callback = None

        if 'error' in kwargs:
            error = kwargs['error']
            del kwargs['error']
        else:
            error = None

        try:
            r = func(*args, **kwargs)
            if not isinstance(r, tuple): r = (r,)
            if callback: self.do_callback(callback, *r)
        except Exception, e:
            if error: self.do_callback(error, e)
            else: print "Unhandled error:", e

    def do_callback(self, callback, *args):
        def _callback(callback, args):
            callback(*args)
            return False

        gobject.idle_add(_callback, callback, args)

We can then inherit this class to provide setup for our specific application:

class WebService(ThreadQueue):
    def __init__(self, guid=None, **kwargs):
        """Initialise the service. If guid is not provided, one will be
           requested (returned in the callback). Pass callback= or error=
           to receive notification of readiness."""
        ThreadQueue.__init__(self)

        self.guid = guid
        self.add_request(self._setup_client, **kwargs)

    def _setup_client(self):
        print "Setting up client"

        ...

        return self.guid

Which we call from our program like this:

class Client(object):
    def __init__(self):
        self.w = WebService(guid=guid, callback=self.client_ready)

    def client_ready(self, guid):
        print "client ready:", guid

gobject.threads_init()

Client()

gtk.main()

What's really cool though is adding methods to the API that are called asynchronously for you. Python makes this possible through the power of decorators. Add the following decorator to a method, and it instead of it being called directly, it will be added to the processing queue.

def async_method(func):
    """Makes the given method asynchronous, meaning when it is called it
       will be queued with add_request.
    """

    def bound_func(obj, *args, **kwargs):
        obj.add_request(func, obj, *args, **kwargs)

    return bound_func

class WebService(ThreadQueue):

    @async_method
    def GetStopInformation(self, stopNo):
        print "Requesting information for stop", stopNo

        ...

And that's it! If you can't follow it, don't worry too much. This is possibly the most Pythonesque bit of code I've ever written, but I've tried to make it generic enough that other people can use it for whatever they need. It's currently part of my app that's beginning to take shape, but the source is here.

Incidently, Maemo people: are there Glade definition files allowing me to use Hildon widgets, GtkBuild and Glade 3? That would be super awesome if there were.