import datetime
import enum
import functools
+import inspect
import logging
import math
import multiprocessing
from typing import Callable, Optional
import warnings
-import thread_utils
+# This module is commonly used by others in here and should avoid
+# taking any unnecessary dependencies back on them.
+import exceptions
+
logger = logging.getLogger(__name__)
# in https://code.google.com/p/verse-quiz/source/browse/trunk/timeout.py
-class TimeoutError(AssertionError):
- def __init__(self, value: str = "Timed Out"):
- self.value = value
-
- def __str__(self):
- return repr(self.value)
-
-
def _raise_exception(exception, error_message: Optional[str]):
if error_message is None:
raise exception()
def timeout(
seconds: float = 1.0,
use_signals: Optional[bool] = None,
- timeout_exception=TimeoutError,
+ timeout_exception=exceptions.TimeoutError,
error_message="Function call timed out",
):
"""Add a timeout parameter to a function and return the function.
parameter. The function is wrapped and returned to the caller.
"""
if use_signals is None:
+ import thread_utils
use_signals = thread_utils.is_current_thread_main_thread()
def decorate(function):
)
return _call_with_sample_rate
return decorator
+
+
+def decorate_matching_methods_with(decorator, acl=None):
+ """Apply decorator to all methods in a class whose names begin with
+ prefix. If prefix is None (default), decorate all methods in the
+ class.
+ """
+ def decorate_the_class(cls):
+ for name, m in inspect.getmembers(cls, inspect.isfunction):
+ if acl is None:
+ setattr(cls, name, decorator(m))
+ else:
+ if acl(name):
+ setattr(cls, name, decorator(m))
+ return cls
+ return decorate_the_class