stouputils.decorators.retrying module#

class RetryPolicy(
exceptions: tuple[type[BaseException], ...],
attempt_limit: int | None,
delays: tuple[float, ...] | None,
delay: float,
backoff: float,
message: str,
on_each_failure: Callable[[BaseException, int], Any] | None,
)[source]#

Bases: object

Helper class for how retry() calls a function again: on which exceptions, how many times, and after which waits.

exceptions: tuple[type[BaseException], ...][source]#

Exceptions to catch and retry on.

attempt_limit: int | None[source]#

Attempts allowed in total, None for no limit.

delays: tuple[float, ...] | None[source]#

Seconds to wait after each failed attempt, replacing delay and backoff when given.

delay: float[source]#

1.0).

Type:

Initial delay in seconds between retries (default

backoff: float[source]#

1.0 for constant delay).

Type:

Multiplier for delay after each retry (default

message: str[source]#

“{ExceptionName} encountered while running {func_name}”).

Type:

Custom message to display before “, retrying” (default

on_each_failure: Callable[[BaseException, int], Any] | None[source]#

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

call(
func: Callable[[...], T],
args: tuple[Any, ...],
kwargs: dict[str, Any],
) → T[source]#

Call func until it returns, raising its last exception once the attempts run out.

wait_after(attempt: int) → float[source]#

Seconds to wait after the given failed attempt, counted from 1.

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]
retry_warning(
error: BaseException,
func: Callable[[...], Any],
message: str,
delay: float,
attempt: int,
attempt_limit: int | None,
) → str[source]#

The warning retry() prints before waiting for the next attempt.

Parameters:

  • message – Text opening the warning, the error and the function name when empty.

  • attempt – Number of the attempt that just failed, from 1.

  • attempt_limit – Attempts allowed in total, None for no limit.

>>> retry_warning(OSError("busy"), print, "", 0.5, 1, None)
'OSError encountered while running print(), retrying in 0.5s (2/∞): busy'
>>> retry_warning(OSError("busy"), print, "Lock taken", 2.0, 2, 3)
'Lock taken, retrying in 2.0s (3/3): busy'