def _raise_exception(exception, error_message: Optional[str]):
if error_message is None:
- raise exception()
+ raise Exception()
else:
- raise exception(error_message)
+ raise Exception(error_message)
def _target(queue, function, *args, **kwargs):
class _Timeout(object):
- """Wrap a function and add a timeout (limit) attribute to it.
+ """Wrap a function and add a timeout to it.
Instances of this class are automatically generated by the add_timeout
- function defined below.
+ function defined below. Do not use directly.
"""
def __init__(
main thread). When not using signals, timeout granularity will be
rounded to the nearest 0.1s.
- Raises an exception when the timeout is reached.
+ Raises an exception when/if the timeout is reached.
It is illegal to pass anything other than a function as the first
parameter. The function is wrapped and returned to the caller.
+
+ >>> @timeout(0.2)
+ ... def foo(delay: float):
+ ... time.sleep(delay)
+ ... return "ok"
+
+ >>> foo(0)
+ 'ok'
+
+ >>> foo(1.0)
+ Traceback (most recent call last):
+ ...
+ Exception: Function call timed out
+
"""
if use_signals is None:
import thread_utils
self._entered = True
f(*args, **kwargs)
self._entered = False
-
return _gatekeeper