Decorator to make a decorated call in a separate thread with timeout (Python recipe) by Dmitry Dvoinikov

ActiveState Code (http://code.activestate.com/recipes/440569/)

#!/usr/bin/env python
# -*- coding: iso-8859-1 -*-
################################################################################
#
# Decorator @threadmethod(sec), makes decorated method calls to always 
# execute in a separate new thread with a specified timeout, propagating 
# exceptions, as well as a result. 
# Dmitry Dvoinikov <dmitry@targeted.org>
#
# from threadmethod import *
#
# class NetworkedSomething(object):
#     @threadmethod(10.0)
#     def connect(self, host, port):
#         ... this could take long long time ...
#    
# # the following call throws ThreadMethodTimeoutError upon a 10 sec. timeout
# NetworkedSomething().connect("123.45.67.89", 1234). Similarly, 
#
# @threadmethod()
# def foo():
#     ...
#
# makes foo() an async method, which just executes in a new separate thread
# each time, but that thread is not waited for, it's just launched to execute 
# in parallel. Besides, in the latter case foo() returns a reference to the
# created thread, so that it can be join()ed.
#
################################################################################

__all__ = [ "threadmethod", "ThreadMethodTimeoutError" ]

################################################################################

class ThreadMethodTimeoutError(Exception): pass

################################################################################

from threading import Thread

class ThreadMethodThread(Thread):
    "ThreadMethodThread, daemonic descendant class of threading.Thread which " \
    "simply runs the specified target method with the specified arguments."

    def __init__(self, target, args, kwargs):
        Thread.__init__(self)
        self.setDaemon(True)
        self.target, self.args, self.kwargs = target, args, kwargs
        self.start()

    def run(self):
        try:
            self.result = self.target(*self.args, **self.kwargs)
        except Exception, e:
            self.exception = e
        except:
            self.exception = Exception()
        else:
            self.exception = None

################################################################################

def threadmethod(timeout = None):
    "@threadmethod(timeout), decorator function, returns a method wrapper " \
    "which runs the wrapped method in a separate new thread."

    def threadmethod_proxy(method):
    
        if hasattr(method, "__name__"):
            method_name = method.__name__
        else:
            method_name = "unknown"

        def threadmethod_invocation_proxy(*args, **kwargs):
            worker = ThreadMethodThread(method, args, kwargs)
            if timeout is None:
                return worker
            worker.join(timeout)
            if worker.isAlive():
                raise ThreadMethodTimeoutError("A call to %s() has timed out" 
                                               % method_name)
            elif worker.exception is not None:
                raise worker.exception
            else:
                return worker.result

        threadmethod_invocation_proxy.__name__ = method_name

        return threadmethod_invocation_proxy

    return threadmethod_proxy

################################################################################

if __name__ == "__main__": # run self-tests

    print "self-testing module threadmethod.py:"

    from threading import currentThread

    mainthread = currentThread()
    @threadmethod(5)
    def tryme():
        assert currentThread() is not mainthread
    tryme()

    @threadmethod(5)
    def foo(a, b, c):
        return a + b + c
    assert foo(1, 2, 3) == 6

    @threadmethod(5)
    def foo(*args):
        assert args == ("foo", )
        return args[0]
    assert foo("foo") == "foo"

    @threadmethod(5)
    def foo(**kwargs):
        assert kwargs == { "foo" : "bar" }
        return kwargs["foo"]
    assert foo(foo = "bar") == "bar"

    @threadmethod(5)
    def foo(a, b, *args, **kwargs):
        assert a == 1 and b == "foo" and args == ("bar", ) and kwargs == { "biz" : "baz" }
    assert foo(1, "foo", "bar", biz = "baz") is None

    from time import sleep
    
    class bar(object):
        @threadmethod(3)
        def __init__(self, timeout):
            sleep(timeout)
        @threadmethod(1)
        def throw(self, e):
            raise e

    try:
        bar(5)
    except ThreadMethodTimeoutError:
        pass
    else:
        assert False, "Constructor should have timed out"

    try:
        bar(1).throw(IOError("fatal"))
    except IOError, e:
        assert str(e) == "fatal"
    else:
        assert False, "Expected IOError(\"fatal\")"

    x = 0

    @threadmethod()
    def async():
        global x
        sleep(0.25)
        x += 1

    async()

    while x == 0:
        pass

    @threadmethod()
    def foo():
        sleep(1.0)
        
    foo().join()

    print "ok"

################################################################################