timeout-decorator
timeout-decorator copied to clipboard
using wrapt instead of wraps, refractoring only use one wrapped function, change Exception, adding powerful eval option
- using wrapt instead of wraps - this give more accurate results when debugging in pydev / docrunner / doctest, especially when decorating class methods. wraps just does not behave completely correct.
- disable the use of signals in windows automatically - they wont work anyway
- the default timeout_exception_message does now also show the function name
- the subclass of TimeoutError is omitted - we want to raise the original exception passed as parameter, not a subclass (a matter of taste) - below Python 3.3 an AssertionError is thrown instead of TimeoutError.
- reduced to one wrapped function - no need for two, avoiding duplicating code
- changed the parameter name to timeout - and to dec_timeout when parameter is passed through the functions kwargs.
- added pypy and pypy3 to the tests
- added allow_eval - very powerful. You can access now attributes of the wrapped function, attributes of the instance of a class, arg, kwargs to get the timeout value.
@timeout(5)
def foo():
interesting stuff happens here
@timeout(dec_timeout=20)
def foo2():
foo(dec_timeout=1) # overriding the timeout of 5 seconds defined before in the decorator
# of foo. I chose the name dec_timeout not to collide with other
# keyword parameters the function might use the parameter name "timeout"
# itself - it would be a too common name ...
# so dec_timeout for decorator timeout.
def foo3(x,y,z):
print('{x}-{y}-{z}'.format(**locals()))
# use the decorator without decorating the function :
timeout(3)(foo3)(1,2,3)
'1-2-3'
# using the dec_allow_eval parameter:
# This is very powerful, but is also very dangerous if you accept strings to evaluate from
# UNTRUSTED input.
# read: https://nedbatchelder.com/blog/201206/eval_really_is_dangerous.html
def class Foo(object):
def __init__(self,x):
self.x=x
@timeout(dec_timeout='instance.x', dec_allow_eval=True)
def foo2(self):
print('swallow')
@timeout(1)
def foo3(self):
print('parrot')
@timeout(dec_timeout='args[0] + kwargs.pop("max_time",0)', dec_allow_eval=True)
def foo4(self,base_delay):
time.sleep(base_delay)
print('knight')
# or override via kwarg :
my_foo = Foo(3)
my_foo.foo2(dec_timeout='instance.x * 2.5 +1')
my_foo.foo3(dec_timeout='instance.x * 2.5 +1', dec_allow_eval=True)
my_foo.foo4(1,max_time=3) # this will time out in 4 seconds
implemented multiprocess.pipe instead of multiprocessing.queue -it is faster and can be probably used on amazon AWS
implemented multiprocess.pipe instead of multiprocessing.queue -it is faster and can be probably used on amazon AWS