import time
import traceback
import warnings
-from typing import Any, Callable, List, Optional
+from typing import Any, Callable, List, NoReturn, Optional, Union
# This module is commonly used by others in here and should avoid
# taking any unnecessary dependencies back on them.
wait_time = min_interval_seconds - elapsed_since_last
else:
wait_time = 0.0
- logger.debug('@%.4f> wait_time = %.4f', time.time(), wait_time)
+ logger.debug("@%.4f> wait_time = %.4f", time.time(), wait_time)
return wait_time
def wrapper_wrapper_rate_limited(*args, **kargs) -> Any:
):
break
with cv:
- logger.debug('@%.4f> calling it...', time.time())
+ logger.debug("@%.4f> calling it...", time.time())
ret = func(*args, **kargs)
last_invocation_timestamp[0] = time.time()
logger.debug(
- '@%.4f> Last invocation <- %.4f',
+ "@%.4f> Last invocation <- %.4f",
time.time(),
last_invocation_timestamp[0],
)
def __call__(self, *args, **kwargs):
"""Returns a single instance of decorated class"""
logger.debug(
- '@singleton returning global instance of %s', self.__wrapped__.__name__
+ "@singleton returning global instance of %s", self.__wrapped__.__name__
)
if self._instance is None:
self._instance = self.__wrapped__(*args, **kwargs)
cache_key = args + tuple(kwargs.items())
if cache_key not in wrapper_memoized.cache:
value = func(*args, **kwargs)
- logger.debug('Memoizing %s => %s for %s', cache_key, value, func.__name__)
+ logger.debug("Memoizing %s => %s for %s", cache_key, value, func.__name__)
wrapper_memoized.cache[cache_key] = value
else:
- logger.debug('Returning memoized value for %s', {func.__name__})
+ logger.debug("Returning memoized value for %s", {func.__name__})
return wrapper_memoized.cache[cache_key]
wrapper_memoized.cache = {} # type: ignore
that we should stop calling or False to indicate a retry
is necessary
+ Raises:
+ ValueError: on invalid arguments; e.g. backoff must be >= 1.0,
+ delay_sec must be >= 0.0, tries must be > 0.
+
.. note::
If after `tries` attempts the wrapped function is still
@functools.wraps(f)
def f_retry(*args, **kwargs):
mtries, mdelay = tries, delay_sec # make mutable
- logger.debug('deco_retry: will make up to %d attempts...', mtries)
+ logger.debug("deco_retry: will make up to %d attempts...", mtries)
retval = f(*args, **kwargs)
while mtries > 0:
if predicate(retval) is True:
- logger.debug('Predicate succeeded, deco_retry is done.')
+ logger.debug("Predicate succeeded, deco_retry is done.")
return retval
logger.debug("Predicate failed, sleeping and retrying.")
mtries -= 1
return deco_retry
-def retry_if_false(tries: int, *, delay_sec=3.0, backoff=2.0):
+def retry_if_false(tries: int, *, delay_sec: float = 3.0, backoff: float = 2.0):
"""A helper for `@predicated_retry_with_backoff` that retries a
decorated function as long as it keeps returning False.
)
-def retry_if_none(tries: int, *, delay_sec=3.0, backoff=2.0):
+def retry_if_none(tries: int, *, delay_sec: float = 3.0, backoff: float = 2.0):
"""A helper for `@predicated_retry_with_backoff` that continues to
invoke the wrapped function as long as it keeps returning None.
Retries up to N times with a delay between each retry and a
# to be bound by the terms and conditions of this License Agreement.
-def _raise_exception(exception, error_message: Optional[str]):
+def _raise_exception(exception, error_message: Optional[str]) -> NoReturn:
"""Internal. Raise a deferred exception"""
if error_message is None:
raise Exception(exception)
`use_signals` argument explicitly.
Raises:
-
- An Exception with a timed out message when/if the timeout is
- reached.
+ Exception: the timeout was reached
It is illegal to pass anything other than a function as the first
parameter. The function is wrapped and returned to the caller.
return decorate
-def synchronized(lock):
+def synchronized(lock: Union[threading.Lock, threading.RLock]):
"""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.
probability_of_call: probability with which to invoke the
wrapped function. Must be 0 <= probabilty <= 1.0.
+ Raises:
+ ValueError: invalid probability argument
+
Example usage... this example would skip the invocation of
`log_the_entire_request_message` 95% of the time and only invoke
if 5% of the time.::
return decorator
-def decorate_matching_methods_with(decorator, acl=None):
+def decorate_matching_methods_with(decorator: Callable, acl: Optional[Callable] = None):
"""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):
for name, m in inspect.getmembers(cls, inspect.isfunction):
- if acl is None:
+ if acl is None or acl(name):
setattr(cls, name, decorator(m))
- else:
- if acl(name):
- setattr(cls, name, decorator(m))
return cls
return decorate_the_class
-if __name__ == '__main__':
+if __name__ == "__main__":
import doctest
doctest.testmod()
projects / pyutils.git / blobdiff