# similar to https://github.com/invl/retry/blob/master/retry/api.py
import functools
import random
import time
from jammy.logging import get_logger
logging_logger = get_logger()
# pylint: disable=too-many-arguments, broad-except
def __retry_internal(
f,
exceptions=Exception,
tries=-1,
delay=0,
max_delay=None,
backoff=1,
jitter=0,
logger=logging_logger,
):
"""
Executes a function and retries it if it failed.
:param f: the function to execute.
:param exceptions: an exception or a tuple of exceptions to catch. default: Exception.
:param tries: the maximum number of attempts. default: -1 (infinite).
:param delay: initial delay between attempts. default: 0.
:param max_delay: the maximum value of delay. default: None (no limit).
:param backoff: multiplier applied to delay between attempts. default: 1 (no backoff).
:param jitter: extra seconds added to delay between attempts. default: 0.
fixed if a number, random if a range tuple (min, max)
:param logger: logger.warning(fmt, error, delay) will be called on failed attempts.
default: retry.logging_logger. if None, logging is disabled.
:returns: the result of the f function.
"""
_tries, _delay = tries, delay
while _tries:
try:
return f()
except exceptions as e:
_tries -= 1
if not _tries:
raise
if logger is not None:
logger.critical(f"{e}, retrying in {_delay} seconds...")
time.sleep(_delay)
_delay *= backoff
if isinstance(jitter, tuple):
_delay += random.uniform(*jitter)
else:
_delay += jitter
if max_delay is not None:
_delay = min(_delay, max_delay)
[docs]def retry_call(
f,
fargs=None,
fkwargs=None,
exceptions=Exception,
tries=-1,
delay=0,
max_delay=None,
backoff=1,
jitter=0,
logger=logging_logger,
):
"""
Calls a function and re-executes it if it failed.
:param f: the function to execute.
:param fargs: the positional arguments of the function to execute.
:param fkwargs: the named arguments of the function to execute.
:param exceptions: an exception or a tuple of exceptions to catch. default: Exception.
:param tries: the maximum number of attempts. default: -1 (infinite).
:param delay: initial delay between attempts. default: 0.
:param max_delay: the maximum value of delay. default: None (no limit).
:param backoff: multiplier applied to delay between attempts. default: 1 (no backoff).
:param jitter: extra seconds added to delay between attempts. default: 0.
fixed if a number, random if a range tuple (min, max)
:param logger: logger.warning(fmt, error, delay) will be called on failed attempts.
default: retry.logging_logger. if None, logging is disabled.
:returns: the result of the f function.
"""
args = fargs if fargs else list()
kwargs = fkwargs if fkwargs else dict()
return __retry_internal(
functools.partial(f, *args, **kwargs),
exceptions,
tries,
delay,
max_delay,
backoff,
jitter,
logger,
)
[docs]def retry(
exceptions=Exception,
tries=-1,
delay=0,
max_delay=None,
backoff=1,
jitter=0,
logger=logging_logger,
):
"""Returns a retry decorator.
:param exceptions: an exception or a tuple of exceptions to catch. default: Exception.
:param tries: the maximum number of attempts. default: -1 (infinite).
:param delay: initial delay between attempts. default: 0.
:param max_delay: the maximum value of delay. default: None (no limit).
:param backoff: multiplier applied to delay between attempts. default: 1 (no backoff).
:param jitter: extra seconds added to delay between attempts. default: 0.
fixed if a number, random if a range tuple (min, max)
:param logger: logger.warning(fmt, error, delay) will be called on failed attempts.
default: retry.logging_logger. if None, logging is disabled.
:returns: a retry decorator.
"""
@functools.wraps
def retry_decorator(f, *fargs, **fkwargs):
args = fargs if fargs else list()
kwargs = fkwargs if fkwargs else dict()
return __retry_internal(
functools.partial(f, *args, **kwargs),
exceptions,
tries,
delay,
max_delay,
backoff,
jitter,
logger,
)
return retry_decorator
