#!/usr/bin/env python3
-"""Decorators."""
+# © Copyright 2021-2022, Scott Gasch
+# Portions (marked) below retain the original author's copyright.
+
+"""Useful(?) decorators."""
import enum
import functools
import time
import traceback
import warnings
-from typing import Any, Callable, Optional
+from typing import Any, Callable, List, Optional
# This module is commonly used by others in here and should avoid
# taking any unnecessary dependencies back on them.
>>> @timed
... def foo():
... import time
- ... time.sleep(0.1)
+ ... time.sleep(0.01)
>>> foo() # doctest: +ELLIPSIS
Finished foo in ...
def invocation_logged(func: Callable) -> Callable:
- """Log the call of a function.
+ """Log the call of a function on stdout and the info log.
>>> @invocation_logged
... def foo():
def rate_limited(n_calls: int, *, per_period_in_seconds: float = 1.0) -> Callable:
- """Limit invocation of a wrapped function to n calls per period.
+ """Limit invocation of a wrapped function to n calls per time period.
Thread safe. In testing this was relatively fair with multiple
- threads using it though that hasn't been measured.
+ threads using it though that hasn't been measured in detail.
>>> import time
>>> import decorator_utils
seconds: float = 1.0,
when: DelayWhen = DelayWhen.BEFORE_CALL,
) -> Callable:
- """Delay the execution of a function by sleeping before and/or after.
-
- Slow down a function by inserting a delay before and/or after its
+ """Slow down a function by inserting a delay before and/or after its
invocation.
>>> import time
"""Keep a cache of previous function call results.
The cache here is a dict with a key based on the arguments to the
- call. Consider also: functools.lru_cache for a more advanced
- implementation.
+ call. Consider also: functools.cache for a more advanced
+ implementation. See:
+ https://docs.python.org/3/library/functools.html#functools.cache
>>> import time
delay_sec: float = 3.0,
backoff: float = 2.0,
):
- """Retries a function or method up to a certain number of times
- with a prescribed initial delay period and backoff rate.
-
- tries is the maximum number of attempts to run the function.
- 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.
-
+ """Retries a function or method up to a certain number of times with a
+ prescribed initial delay period and backoff rate (multiplier).
+
+ Args:
+ tries: the maximum number of attempts to run the function
+ delay_sec: sets the initial delay period in seconds
+ backoff: a multiplier (must be >=1.0) used to modify the
+ delay at each subsequent invocation
+ predicate: a Callable that will be passed the retval of
+ the decorated function and must return True to indicate
+ that we should stop calling or False to indicate a retry
+ is necessary
"""
+
if backoff < 1.0:
msg = f"backoff must be greater than or equal to 1, got {backoff}"
logger.critical(msg)
>>> import time
>>> counter = 0
-
>>> @retry_if_false(5, delay_sec=1.0, backoff=1.1)
... def foo():
... global counter
"""Another helper for @retry_predicate above. Retries up to N
times so long as the wrapped function returns None with a delay
between each retry and a backoff that can increase the delay.
-
"""
+
return retry_predicate(
tries,
predicate=lambda x: x is not None,
"""This is a decorator which can be used to mark functions
as deprecated. It will result in a warning being emitted
when the function is used.
-
"""
@functools.wraps(func)
wait_event = threading.Event()
result = [None]
- exc = [False, None]
+ exc: List[Any] = [False, None]
def worker_func():
try:
def thunk():
wait_event.wait()
if exc[0]:
+ assert exc[1]
raise exc[1][0](exc[1][1])
return result[0]
main thread). When not using signals, timeout granularity will be
rounded to the nearest 0.1s.
+ Beware that an @timeout on a function inside a module will be
+ evaluated at module load time and not when the wrapped function is
+ invoked. This can lead to problems when relying on the automatic
+ main thread detection code (use_signals=None, the default) since
+ the import probably happens on the main thread and the invocation
+ can happen on a different thread (which can't use signals).
+
Raises an exception when/if the timeout is reached.
It is illegal to pass anything other than a function as the first
def decorate(function):
if use_signals:
- def handler(signum, frame):
+ def handler(unused_signum, unused_frame):
_raise_exception(timeout_exception, error_message)
@functools.wraps(function)
def synchronized(lock):
+ """Emulates java's synchronized keyword: given a lock, require that
+ threads take that lock (or wait) before invoking the wrapped
+ function and automatically releases the lock afterwards.
+ """
+
def wrap(f):
@functools.wraps(f)
def _gatekeeper(*args, **kw):
def call_with_sample_rate(sample_rate: float) -> Callable:
+ """Calls the wrapped function probabilistically given a rate between
+ 0.0 and 1.0 inclusive (0% probability and 100% probability).
+ """
+
if not 0.0 <= sample_rate <= 1.0:
msg = f"sample_rate must be between [0, 1]. Got {sample_rate}."
logger.critical(msg)
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.
+ """Apply the given 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):
projects / python_utils.git / blobdiff