stouputils.decorators.retry module#

retry(
func: Callable[..., T],
*,
exceptions: tuple[type[BaseException], ...] | type[BaseException] = (Exception,),
max_attempts: int | 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 | 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 (Callable[..., T] | None) – Function to retry

  • exceptions (tuple[type[BaseException], ...]) – Exceptions to catch and retry on

  • max_attempts (int | None) – Maximum number of attempts (None for infinite retries)

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

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

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

  • on_each_failure (Callable[[BaseException, int], Any] | None) – 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

Return type:

Callable[…, T]

Examples

>>> 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

>>> # Example: 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)]