stouputils.decorators.retrying module#

retry(
func: Callable[..., T],
*,
exceptions: tuple[type[BaseException], ...] | type[BaseException] = (Exception,),
max_attempts: int | Iterable[float] | None = 10,
delay: float = 1.0,
backoff: float = 1.0,
message: str = '',
on_each_failure: Callable[[BaseException, int], Any] | None = None,
) → Callable[..., T][source]#
retry(
func: None = None,
*,
exceptions: tuple[type[BaseException], ...] | type[BaseException] = (Exception,),
max_attempts: int | Iterable[float] | None = 10,
delay: float = 1.0,
backoff: float = 1.0,
message: str = '',
on_each_failure: Callable[[BaseException, int], Any] | None = None,
) → Callable[[Callable[..., T]], Callable[..., T]]

Decorator that retries a function when specific exceptions are raised.

Parameters:

  • func – Function to retry

  • exceptions – Exceptions to catch and retry on

  • max_attempts – Maximum number of attempts, None for infinite retries. An iterable of seconds gives one delay per attempt instead: its length is the attempt count, and delay/backoff are ignored.

  • delay – Initial delay in seconds between retries (default: 1.0)

  • backoff – Multiplier for delay after each retry (default: 1.0 for constant delay)

  • message – Custom message to display before “, retrying” (default: “{ExceptionName} encountered while running {func_name}”)

  • on_each_failure – Optional callback function to call on each failure, receives the exception and the attempt number as arguments

Returns:

Decorator that retries the function on specified exceptions

>>> import os
>>> @retry(exceptions=PermissionError, max_attempts=3, delay=0.1)
... def write_file():
...     with open("test.txt", "w") as f:
...         f.write("test")

>>> @retry(exceptions=(OSError, IOError), delay=0.5, backoff=2.0)
... def network_call():
...     pass

>>> @retry(max_attempts=5, delay=1.0)
... def might_fail():
...     pass

>>> # Use a lambda to record attempts on each failure
>>> calls = []
>>> @retry(max_attempts=3, delay=0.0, on_each_failure=lambda e, a: calls.append((e, a)))
... def will_fail():
...     raise RuntimeError("nope")
>>> try:
...     will_fail()
... except RuntimeError:
...     pass
>>> calls
[(RuntimeError('nope'), 1), (RuntimeError('nope'), 2), (RuntimeError('nope'), 3)]

>>> # An iterable of delays sets both the waits and the attempt count
>>> attempts = []
>>> @retry(max_attempts=(0.0, 0.0, 0.0, 0.0), on_each_failure=lambda e, a: attempts.append(a))
... def flaky():
...     raise ValueError("nope")
>>> try:
...     flaky()
... except ValueError:
...     pass
>>> attempts
[1, 2, 3, 4]