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__)
@functools.wraps(func)
def wrapper_invocation_logged(*args, **kwargs):
- now = datetime.datetime.now()
- ts = now.strftime("%Y/%d/%b:%H:%M:%S%Z")
- msg = f"[{ts}]: Entered {func.__name__}"
+ msg = f"Entered {func.__qualname__}"
print(msg)
logger.info(msg)
ret = func(*args, **kwargs)
- now = datetime.datetime.now()
- ts = now.strftime("%Y/%d/%b:%H:%M:%S%Z")
- msg = f"[{ts}]: Exited {func.__name__}"
+ msg = f"Exited {func.__qualname__}"
print(msg)
logger.info(msg)
return ret
tries: int,
*,
predicate: Callable[..., bool],
- delay_sec: float = 3,
+ delay_sec: float = 3.0,
backoff: float = 2.0,
):
"""Retries a function or method up to a certain number of times
delay_sec sets the initial delay period in seconds.
backoff is a multiplied (must be >1) used to modify the delay.
predicate is a function that will be passed the retval of the
- decorated function and must return True to stop or False to
- retry.
+ decorated function and must return True to stop or False to
+ retry.
"""
- if backoff < 1:
+ if backoff < 1.0:
msg = f"backoff must be greater than or equal to 1, got {backoff}"
logger.critical(msg)
raise ValueError(msg)
@functools.wraps(f)
def f_retry(*args, **kwargs):
mtries, mdelay = tries, delay_sec # make mutable
+ logger.debug(f'deco_retry: will make up to {mtries} attempts...')
retval = f(*args, **kwargs)
while mtries > 0:
if predicate(retval) is True:
+ logger.debug('Predicate succeeded, deco_retry is done.')
return retval
logger.debug("Predicate failed, sleeping and retrying.")
mtries -= 1
exc[0] = True
exc[1] = sys.exc_info() # (type, value, traceback)
msg = f"Thunkify has thrown an exception (will be raised on thunk()):\n{traceback.format_exc()}"
- logger.warning(msg)
print(msg)
+ logger.warning(msg)
finally:
wait_event.set()
# 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()
"""
try:
queue.put((True, function(*args, **kwargs)))
- except:
+ except Exception:
queue.put((False, sys.exc_info()[1]))
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):
-
if use_signals:
def handler(signum, frame):
)
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