stopit 1.0.0

Using stopit in your application

Both work identically:

easy_install stopit
pip install stopit

Developing stopit

git clone https://github.com/glenfant/stopit.git
cd stopit
python setup.py develop

# Does it work for you ?
python setup.py test

async_raise function raises an exception in another thread

Testing async_raise() with a thread of 5 seconds

>>> five_seconds_threads = threading.Thread(
...     target=variable_duration_func_handling_exc, args=(5.0, exc_traces))
>>> start_time = time.time()
>>> five_seconds_threads.start()
>>> thread_ident = five_seconds_threads.ident
>>> five_seconds_threads.is_alive()
True

We raise a LookupError in that thread

>>> async_raise(thread_ident, LookupError)

Okay but we must wait few milliseconds the thread death since the exception is asynchronous

>>> while five_seconds_threads.is_alive():
...     pass

And we can notice that we stopped the thread before it stopped by itself

>>> time.time() - start_time < 0.5
True
>>> len(exc_traces)
1
>>> exc_traces[-1].__class__.__name__
'LookupError'

Timeout context manager

The context manager stops the execution of its inner block after a given time. You may manage the way the timeout occurs using a try: ... except: ... construct or by inspecting the context manager state attribute after the block.

>>> with Timeout(5.0) as timeout_ctx:
...     result = fast_func()
>>> result
0
>>> timeout_ctx.state == timeout_ctx.EXECUTED
True

>>> start_time = time.time()
>>> with Timeout(2.0) as timeout_ctx:
...     variable_duration_func(5.0)
>>> time.time() - start_time < 2.1
True
>>> timeout_ctx.state == timeout_ctx.TIMED_OUT
True

>>> try:
...     with Timeout(5.0) as timeout_ctx:
...         result = func_with_exception()
... except LookupError:
...     result = 'exception_seen'
>>> timeout_ctx.state == timeout_ctx.EXECUTING
True
>>> result
'exception_seen'

>>> result = None
>>> start_time = time.time()
>>> try:
...     with Timeout(2.0, swallow_exc=False) as timeout_ctx:
...         variable_duration_func(5.0)
... except TimeoutException:
...     result = 'exception_seen'
>>> time.time() - start_time < 2.1
True
>>> result
'exception_seen'
>>> timeout_ctx.state == timeout_ctx.TIMED_OUT
True

>>> result = None
>>> start_time = time.time()
>>> try:
...     with Timeout(2.0, swallow_exc=False) as timeout_ctx:
...         func_with_exception()
... except Exception:
...     result = 'exception_seen'
>>> time.time() - start_time < 0.1
True
>>> result
'exception_seen'
>>> timeout_ctx.state == timeout_ctx.EXECUTING
True

timeoutable callable decorator

This decorator stops the execution of any callable that should not last a certain amount of time.

You may use a decorated callable without timeout control if you don’t provide the timeout optionl argument:

>>> @timeoutable()
... def fast_double(value):
...     return value * 2
>>> fast_double(3)
6

You may specify that timeout with the timeout optional argument. Interrupted callables return None:

>>> @timeoutable()
... def infinite():
...     while True:
...         pass
...     return 'whatever'
>>> infinite(timeout=1) is None
True

Or any other value provided to the timeoutable decorator parameter:

>>> @timeoutable('unexpected')
... def infinite():
...     while True:
...         pass
...     return 'whatever'
>>> infinite(timeout=1)
'unexpected'

If the timeout parameter name may clash with your callable signature, you may change it using timeout_param:

>>> @timeoutable('unexpected', timeout_param='my_timeout')
... def infinite():
...     while True:
...         pass
...     return 'whatever'
>>> infinite(my_timeout=1)
'unexpected'

It works on instance methods too:

>>> class Anything(object):
...     @timeoutable('unexpected')
...     def infinite(self, value):
...         assert type(value) is int
...         while True:
...             pass
>>> obj = Anything()
>>> obj.infinite(2, timeout=1)
'unexpected'