Async like pattern in pyqt? Or cleaner background call pattern?

Question

I'm trying to write a short(one file pyqt) program which is responsive(so dependencies outside python/lxml/qt, especially ones I can't just stick in the file have some downsides for this use case but I might still be willing to try them). I'm trying to perform possibly lengthy(and cancelable) operations on a worker thread(actually the background operation has a lock around it to prevent multiple operations at once(since the library it uses can only be used one call at a time) and timeouts so spawning multiple threads would be fine also).

As far as I can figure out the "basic" way to do this with qt is. (note code is not tested so it may be wrong)

class MainWindow(QWidget):
    #self.worker moved to background thread
    def initUI(self):
        ...
        self.cmd_button.clicked.connect(self.send)
        ...

    @pyqtslot()
    def send(self):
        ...
        ...#get cmd from gui
        QtCore.QTimer.singleShot(0, lambda : self.worker(cmd))


    @pyqtslot(str)
    def end_send(self, result):
        ...
        ...# set some gui to display result
        ...



class WorkerObject(QObject):    
   def send_cmd(self, cmd):
       ... get result of cmd
       QtCore.QTimer.singleShot(0, lambda: self.main_window.end_send())

(Am I using QTimer right(it runs on different thread right)?)

I'd really prefer to have something simpler and more abstracted along the lines of c#'s async. (note I haven't used asyncio so I might be getting some things wrong)

class MainWindow(QWidget):
    ...
    @asyncio.coroutine
    def send(self):
        ...
        ...#get cmd from gui
        result = yield from self.worker(cmd)
        #set gui textbox to result

class WorkerObject(QObject):
   @asyncio.coroutine
   def send_cmd(self, cmd):
       ... get result of cmd
       yield from loop.run_in_executor(None, self.model.send_command, cmd)

I heard that python 3 had similar features and there was a back port but does it work properly with qt?

If anyone knows of another saner pattern. that too would be useful/an acceptable answer.

Answer 1

The short answer to your question ("is there a way to use an asyncio-like pattern in PyQt?") is yes, but it's pretty complicated and arguably not worth it for a small program. Here's some prototype code that allows you to use an asynchronous pattern like you described:

import types
import weakref
from functools import partial

from PyQt4 import QtGui 
from PyQt4 import QtCore
from PyQt4.QtCore import QThread, QTimer

## The following code is borrowed from here: 
# http://stackoverflow.com/questions/24689800/async-like-pattern-in-pyqt-or-cleaner-background-call-pattern
# It provides a child->parent thread-communication mechanism.
class ref(object):
    """
    A weak method implementation
    """
    def __init__(self, method):
        try:
            if method.im_self is not None:
                # bound method
                self._obj = weakref.ref(method.im_self)
            else:
                # unbound method
                self._obj = None
            self._func = method.im_func
            self._class = method.im_class
        except AttributeError:
            # not a method
            self._obj = None
            self._func = method
            self._class = None

    def __call__(self):
        """
        Return a new bound-method like the original, or the
        original function if refers just to a function or unbound
        method.
        Returns None if the original object doesn't exist
        """
        if self.is_dead():
            return None
        if self._obj is not None:
            # we have an instance: return a bound method
            return types.MethodType(self._func, self._obj(), self._class)
        else:
            # we don't have an instance: return just the function
            return self._func

    def is_dead(self):
        """
        Returns True if the referenced callable was a bound method and
        the instance no longer exists. Otherwise, return False.
        """
        return self._obj is not None and self._obj() is None

    def __eq__(self, other):
        try:
            return type(self) is type(other) and self() == other()
        except:
            return False

    def __ne__(self, other):
        return not self == other

class proxy(ref):
    """
    Exactly like ref, but calling it will cause the referent method to
    be called with the same arguments. If the referent's object no longer lives,
    ReferenceError is raised.

    If quiet is True, then a ReferenceError is not raise and the callback 
    silently fails if it is no longer valid. 
    """

    def __init__(self, method, quiet=False):
        super(proxy, self).__init__(method)
        self._quiet = quiet

    def __call__(self, *args, **kwargs):
        func = ref.__call__(self)
        if func is None:
            if self._quiet:
                return
            else:
                raise ReferenceError('object is dead')
        else:
            return func(*args, **kwargs)

    def __eq__(self, other):
        try:
            func1 = ref.__call__(self)
            func2 = ref.__call__(other)
            return type(self) == type(other) and func1 == func2
        except:
            return False

class CallbackEvent(QtCore.QEvent):
    """
    A custom QEvent that contains a callback reference

    Also provides class methods for conveniently executing 
    arbitrary callback, to be dispatched to the event loop.
    """
    EVENT_TYPE = QtCore.QEvent.Type(QtCore.QEvent.registerEventType())

    def __init__(self, func, *args, **kwargs):
        super(CallbackEvent, self).__init__(self.EVENT_TYPE)
        self.func = func
        self.args = args
        self.kwargs = kwargs

    def callback(self):
        """
        Convenience method to run the callable. 

        Equivalent to:  
            self.func(*self.args, **self.kwargs)
        """
        self.func(*self.args, **self.kwargs)

    @classmethod
    def post_to(cls, receiver, func, *args, **kwargs):
        """
        Post a callable to be delivered to a specific
        receiver as a CallbackEvent. 

        It is the responsibility of this receiver to 
        handle the event and choose to call the callback.
        """
        # We can create a weak proxy reference to the
        # callback so that if the object associated with
        # a bound method is deleted, it won't call a dead method
        if not isinstance(func, proxy):
            reference = proxy(func, quiet=True)
        else:
            reference = func
        event = cls(reference, *args, **kwargs)

        # post the event to the given receiver
        QtGui.QApplication.postEvent(receiver, event)

## End borrowed code

## Begin Coroutine-framework code

class AsyncTask(QtCore.QObject):
    """ Object used to manage asynchronous tasks.

    This object should wrap any function that you want
    to call asynchronously. It will launch the function
    in a new thread, and register a listener so that
    `on_finished` is called when the thread is complete.

    """
    def __init__(self, func, *args, **kwargs):
        super(AsyncTask, self).__init__()
        self.result = None  # Used for the result of the thread.
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.finished = False
        self.finished_cb_ran = False
        self.finished_callback = None
        self.objThread = RunThreadCallback(self, self.func, self.on_finished, 
                                           *self.args, **self.kwargs)
        self.objThread.start()

    def customEvent(self, event):
        event.callback()

    def on_finished(self, result):
        """ Called when the threaded operation is complete.

        Saves the result of the thread, and
        executes finished_callback with the result if one
        exists. Also closes/cleans up the thread.

        """
        self.finished = True
        self.result = result
        if self.finished_callback:
            self.finished_ran = True
            func = partial(self.finished_callback, result)
            QTimer.singleShot(0, func)
        self.objThread.quit()
        self.objThread.wait()

class RunThreadCallback(QtCore.QThread):
    """ Runs a function in a thread, and alerts the parent when done. 

    Uses a custom QEvent to alert the main thread of completion.

    """
    def __init__(self, parent, func, on_finish, *args, **kwargs):
        super(RunThreadCallback, self).__init__(parent)
        self.on_finished = on_finish
        self.func = func
        self.args = args
        self.kwargs = kwargs

    def run(self):
        try:
            result = self.func(*self.args, **self.kwargs)
        except Exception as e:
            print "e is %s" % e
            result = e
        finally:
            CallbackEvent.post_to(self.parent(), self.on_finished, result)


def coroutine(func):
    """ Coroutine decorator, meant for use with AsyncTask.

    This decorator must be used on any function that uses
    the `yield AsyncTask(...)` pattern. It shouldn't be used
    in any other case.

    The decorator will yield AsyncTask objects from the
    decorated generator function, and register itself to
    be called when the task is complete. It will also
    excplicitly call itself if the task is already
    complete when it yields it.

    """
    def wrapper(*args, **kwargs):
        def execute(gen, input=None):
            if isinstance(gen, types.GeneratorType):
                if not input:
                    obj = next(gen)
                else:
                    try:
                        obj = gen.send(input)
                    except StopIteration as e:
                        result = getattr(e, "value", None)
                        return result
                if isinstance(obj, AsyncTask):
                    # Tell the thread to call `execute` when its done
                    # using the current generator object.
                    func = partial(execute, gen)
                    obj.finished_callback = func
                    if obj.finished and not obj.finished_cb_ran:
                        obj.on_finished(obj.result)
                else:
                    raise Exception("Using yield is only supported with AsyncTasks.")
            else:
                print("result is %s" % result)
                return result
        result = func(*args, **kwargs)
        execute(result)
    return wrapper

## End coroutine-framework code

If you put the above code into a module (say qtasync.py) you can import it into a script and use it like so to get asyncio-like behavior:

import sys
import time
from qtasync import AsyncTask, coroutine
from PyQt4 import QtGui
from PyQt4.QtCore import QThread

class MainWindow(QtGui.QMainWindow):
    def __init__(self):
        super(MainWindow, self).__init__()
        self.initUI()

    def initUI(self):
        self.cmd_button = QtGui.QPushButton("Push", self)
        self.cmd_button.clicked.connect(self.send_evt)
        self.statusBar()
        self.show()

    def worker(self, inval):
        print "in worker, received '%s'" % inval
        time.sleep(2)
        return "%s worked" % inval

    @coroutine
    def send_evt(self, arg):
        out = AsyncTask(self.worker, "test string")
        out2 = AsyncTask(self.worker, "another test string")
        QThread.sleep(3)
        print("kicked off async task, waiting for it to be done")
        val = yield out
        val2 = yield out2
        print ("out is %s" % val)
        print ("out2 is %s" % val2)
        out = yield AsyncTask(self.worker, "Some other string")
        print ("out is %s" % out)


if __name__ == "__main__":
    app = QtGui.QApplication(sys.argv)
    m = MainWindow()
    sys.exit(app.exec_())

Output (when the button is pushed):

in worker, received 'test string'
in worker, received 'another test string'
kicked off async task, waiting for it to be done
out is test string worked
out2 is another test string worked
in worker, received 'Some other string'
out is Some other string worked

As you can see, worker gets run asynchronously in a thread whenever it gets called via the AsyncTask class, but its return value can be yielded directly from send_evt, without needing to use callbacks.

The code uses the coroutine-supporting features (generator_object.send) of Python generators, and a recipe I found on ActiveState that provides a child->main thread communication mechanism, to implement some very basic coroutines. The coroutines are quite limited: You can't return anything from them, and you can't chain coroutine calls together. It's probably possible to implement both of those things, but also probably not worth the effort, unless you really need them. I haven't done much negative testing with this either, so exceptions in workers and elsewhere may not be handled properly. What it does do well, though, is allow you to call methods in separate threads via the AsyncTask class, and then yield a result from the thread when one is ready, without blocking the Qt event loop. Normally this kind of thing would be done with callbacks, which can be difficult to follow and is generally less readable than having all the code in a single function.

You're welcome to use this approach if the limitations I mentioned are acceptable to you, but this is really just a proof-of-concept; you would need to do a whole bunch of testing before you think about put it into production anywhere.

As you mentioned, Python 3.3 and 3.4 makes asynchronous programming easier with the introduction of yield from and asyncio, respectively. I think yield from would actually be quite useful here to allow chaining coroutines (meaning have one coroutine call another and get a result from it). asyncio has no PyQt4 event-loop integration, so it's usefulness is pretty limited.

Another option would be to drop the coroutine piece of this altogether and just use the callback-based inter-thread communication mechanism directly:

import sys
import time

from qtasync import CallbackEvent  # No need for the coroutine stuff
from PyQt4 import QtGui
from PyQt4.QtCore import QThread

class MyThread(QThread):
    """ Runs a function in a thread, and alerts the parent when done. 

    Uses a custom QEvent to alert the main thread of completion.

    """
    def __init__(self, parent, func, on_finish, *args, **kwargs):
        super(MyThread, self).__init__(parent)
        self.on_finished = on_finish
        self.func = func
        self.args = args
        self.kwargs = kwargs
        self.start()

    def run(self):
        try:
            result = self.func(*self.args, **self.kwargs)
        except Exception as e:
            print "e is %s" % e
            result = e
        finally:
            CallbackEvent.post_to(self.parent(), self.on_finished, result)


class MainWindow(QtGui.QMainWindow):
    def __init__(self):
        super(MainWindow, self).__init__()
        self.initUI()

    def initUI(self):
        self.cmd_button = QtGui.QPushButton("Push", self)
        self.cmd_button.clicked.connect(self.send)
        self.statusBar()
        self.show()

    def customEvent(self, event):
        event.callback()

    def worker(self, inval):
        print("in worker, received '%s'" % inval)
        time.sleep(2)
        return "%s worked" % inval

    def end_send(self, cmd):
        print("send returned '%s'" % cmd)

    def send(self, arg):
        t = MyThread(self, self.worker, self.end_send, "some val")
        print("Kicked off thread")


if __name__ == "__main__":
    app = QtGui.QApplication(sys.argv)
    m = MainWindow()
    sys.exit(app.exec_())

Output:

Kicked off thread
in worker, received 'some val'
send returned 'some val worked'

This could get a bit unwieldy if you're dealing with a long callback chain, but it doesn't rely on the more unproven coroutine code.