Retry and fallback

retry ¶

Retry, fallback and suppress decorators.

RetryState `dataclass` ¶

RetryState(attempt: int, exception: BaseException | None, next_delay: float, elapsed: float, result: Any = None)

Snapshot passed to the on_retry callback before each wait.

Attributes:

Name	Type	Description
`attempt`	`int`	Number of the attempt that just failed (1-based).
`exception`	`BaseException \| None`	The exception that attempt raised, or `None` when the attempt returned a value that `retry_if_result` rejected.
`next_delay`	`float`	Seconds about to be slept before the next attempt (jitter included).
`elapsed`	`float`	Seconds since the first attempt started.
`result`	`Any`	The rejected return value when `exception` is `None`, else `None`.

RetryStats ¶

RetryStats()

Thread-safe counters for one retried function. Read them with retry_stats.

Attributes:

Name	Type	Description
`calls`		Times the wrapped function was called (each call may make several attempts).
`attempts_total`		Individual attempts made across all calls.
`retries`		Times an attempt failed and another one was scheduled.
`gave_up`		Calls that ended by re-raising a retryable exception, or by returning a result `retry_if_result` rejected (attempts, time budget, retry budget or `retry_if` said stop).
`budget_exhausted`		Calls among `gave_up` that stopped because the shared `RetryBudget` refused another retry.
`total_sleep`		Seconds spent waiting between attempts.
`last_exception`	`BaseException \| None`	The most recent exception a retried attempt raised, if any.

snapshot ¶

snapshot() -> dict[str, Any]

Return a consistent copy of all counters as a dict.

reset ¶

reset() -> None

Set every counter back to zero.

RetryBudgetStats `dataclass` ¶

RetryBudgetStats(requests: int, retries: int, rejected: int, allowed: int)

Snapshot of a RetryBudget, returned by RetryBudget.stats.

Attributes:

Name	Type	Description
`requests`	`int`	First attempts recorded in the current window.
`retries`	`int`	Retries granted in the current window.
`rejected`	`int`	Retries refused since creation (or the last `reset`).
`allowed`	`int`	Retries the budget would still grant right now.

RetryBudget ¶

RetryBudget(ratio: float = 0.1, min_per_second: float = 10, window: float = 10.0)

A retry allowance shared by many retried functions (Finagle/Envoy style).

Every call's first attempt deposits into the budget and every retry withdraws from it. A retry is allowed while the retries made in the last window seconds stay within ratio times the requests made in that window, plus a reserve of min_per_second * window retries so that low-traffic code can still retry. When a dependency fails for everyone, retries are capped at about ratio of the traffic instead of multiplying it by max_attempts.

One budget can be passed to any number of retry decorators and retrying blocks; it is thread-safe and works across threads and event loops (it never blocks). Time comes from python_wrap._core.monotonic, so fake_clock controls it in tests.

Parameters:

Name	Type	Description	Default
`ratio`	`float`	Retries allowed per request in the window (`0.1` = 10 %). `>= 0`.	`0.1`
`min_per_second`	`float`	Retries per second always allowed regardless of traffic. `>= 0`.	`10`
`window`	`float`	Length of the sliding window in seconds. `> 0`.	`10.0`

Raises:

Type	Description
`ValueError`	If an argument is out of range.

Example

from python_wrap.retry import RetryBudget, retry

budget = RetryBudget(ratio=0.2, min_per_second=1, window=10)

@retry(max_attempts=3, delay=0, budget=budget)
def fetch() -> str:
    return "ok"

assert fetch() == "ok"
assert budget.stats.requests == 1

stats `property` ¶

stats: RetryBudgetStats

A consistent RetryBudgetStats snapshot of the current window.

deposit ¶

deposit() -> None

Record one request (a first attempt). retry calls this for you.

try_withdraw ¶

try_withdraw() -> bool

Take one retry from the budget.

retry calls this before each retry; call it yourself only in a hand-written loop.

Returns:

Type	Description
`bool`	`True` if the retry is allowed (and was recorded), `False` if the budget is
`bool`	exhausted (recorded as rejected).

reset ¶

reset() -> None

Forget all recorded requests, retries and rejections.

Attempt ¶

Attempt(number: int, run: _Run)

One attempt of a retrying loop; use it as a context manager.

Exceptions that retry would retry are swallowed by with attempt: (the loop then waits and continues); other exceptions, and the failure of the final attempt, propagate. In an async for loop, use async with attempt: when retry_if, retry_if_result or on_retry are async callables (plain with works otherwise).

With retry_if_result, store the value the block produced in attempt.result; the predicate is checked when the block ends and a rejected value starts another attempt. When no retry is left the loop simply ends, leaving the last value in your variable (and in attempt.result).

Attributes:

Name	Type	Description
`number`		1-based number of this attempt.

result `property` `writable` ¶

result: Any

The value checked by retry_if_result (None until you set it).

Retrying ¶

Retrying(engine: _Engine)

Iterable of Attempt objects returned by retrying.

Iterate it with for (sync, sleeps with time.sleep) or async for (sleeps with asyncio). Each iteration over a Retrying is an independent run.

stats `property` ¶

stats: RetryStats

Counters accumulated over all runs of this object.

retry_stats ¶

retry_stats(func: Callable[..., Any]) -> RetryStats | None

Return the RetryStats of a function decorated with retry.

Works through other decorators that use functools.wraps and on bound methods (the counters are shared by all instances of the class).

Parameters:

Name	Type	Description	Default
`func`	`Callable[..., Any]`	A function wrapped by `retry`.	required

Returns:

Type	Description
`RetryStats \| None`	Its statistics, or `None` if `func` is not retried.

Example

@retry ... def f() -> None: ... retry_stats(f).calls 0

retry_after_header ¶

retry_after_header(exc: BaseException) -> float | None

Extract a server-requested wait from an exception, for delay_from=.

Looks at exc.retry_after first, then at exc.response.headers["Retry-After"] (header names are matched case-insensitively). Values may be seconds or an HTTP-date; dates are converted to seconds from now. Works with httpx, requests and aiohttp-style exceptions without importing them, and never raises.

Parameters:

Name	Type	Description	Default
`exc`	`BaseException`	Any exception.	required

Returns:

Type	Description
`float \| None`	Seconds to wait (never negative), or `None` if nothing usable was found.

Example

class Err(Exception): ... retry_after = 3 retry_after_header(Err()) 3.0

retry ¶

retry(func: F, /, *, max_attempts: int | None = ..., delay: float = ..., backoff: float = ..., max_delay: float | None = ..., jitter: float = ..., max_time: float | None = ..., on: ExceptionTypes = ..., retry_if: ExceptionPredicate | None = ..., retry_if_result: ResultPredicate | None = ..., on_retry: RetryCallback | None = ..., delay_from: Callable[[BaseException], float | None] | None = ..., budget: RetryBudget | None = ..., logger: LoggerLike = ..., log_level: int | None = ..., log_exc_info: bool = ...) -> F

retry(func: None = None, /, *, max_attempts: int | None = ..., delay: float = ..., backoff: float = ..., max_delay: float | None = ..., jitter: float = ..., max_time: float | None = ..., on: ExceptionTypes = ..., retry_if: ExceptionPredicate | None = ..., retry_if_result: ResultPredicate | None = ..., on_retry: RetryCallback | None = ..., delay_from: Callable[[BaseException], float | None] | None = ..., budget: RetryBudget | None = ..., logger: LoggerLike = ..., log_level: int | None = ..., log_exc_info: bool = ...) -> Callable[[F], F]

retry(func: Callable[..., Any] | None = None, /, *, max_attempts: int | None = _UNSET, delay: float = _UNSET, backoff: float = _UNSET, max_delay: float | None = _UNSET, jitter: float = _UNSET, max_time: float | None = _UNSET, on: ExceptionTypes = Exception, retry_if: ExceptionPredicate | None = None, retry_if_result: ResultPredicate | None = None, on_retry: RetryCallback | None = None, delay_from: Callable[[BaseException], float | None] | None = None, budget: RetryBudget | None = _UNSET, logger: LoggerLike = None, log_level: int | None = _UNSET, log_exc_info: bool = False) -> Any

Retry a function when it raises (or returns a bad value), with optional backoff.

Works bare (@retry), with options (@retry(max_attempts=5)) and as a plain call (retry(func, max_attempts=5)). Async functions are awaited and slept with asyncio so the event loop is never blocked.

The wait before attempt n + 1 is min(delay * backoff ** (n - 1), max_delay) + random.uniform(0, jitter). The last exception is re-raised unchanged, with its original traceback, once attempts, the time budget or the retry budget are exhausted.

With retry_if_result, a returned value for which the predicate is true is treated as a failure and retried. When no retry is left, that last value is returned (no exception is raised), so check it if it matters.

For async def functions, retry_if, retry_if_result and on_retry may be async def callables; they are awaited. A sync function given an async callable raises TypeError when decorated.

asyncio.CancelledError, KeyboardInterrupt, SystemExit and GeneratorExit are never retried, even with on=BaseException or when listed in on. Cancelling an async call during the backoff sleep cancels it immediately.

Arguments you leave out fall back to python_wrap.config.configure(retry={...}) (read when the decorator is applied), then to the values shown here. Inside a python_wrap.deadline block, it never sleeps past the time remaining: if the next wait would outlast the deadline the last exception is re-raised instead.

Statistics are available through retry_stats. The events retry.attempt_failed, retry.succeeded, retry.gave_up and retry.budget_exhausted are emitted for python_wrap.events. Under python_wrap.testing, an overrides.max_attempts cap lowers max_attempts.

Parameters:

Name	Type	Description	Default
`func`	`Callable[..., Any] \| None`	The function to wrap (when used without parentheses).	`None`
`max_attempts`	`int \| None`	Total number of calls allowed, or `None` for no limit.	`_UNSET`
`delay`	`float`	Base wait in seconds before the second attempt.	`_UNSET`
`backoff`	`float`	Multiplier applied to the wait after each failure (>= 1).	`_UNSET`
`max_delay`	`float \| None`	Upper bound for the backoff wait (before jitter is added); also caps waits returned by `delay_from`.	`_UNSET`
`jitter`	`float`	Up to this many random extra seconds are added to each wait.	`_UNSET`
`max_time`	`float \| None`	Total time budget in seconds, measured from the first attempt. If the next wait would run past it, the last exception is re-raised instead.	`_UNSET`
`on`	`ExceptionTypes`	Exception type(s) that trigger a retry; anything else propagates at once.	`Exception`
`retry_if`	`ExceptionPredicate \| None`	Extra predicate on the exception; it is retried only if this returns True. May be async for async functions.	`None`
`retry_if_result`	`ResultPredicate \| None`	Predicate on the return value; True means "retry". When no retry is left the last value is returned. May be async for async functions.	`None`
`on_retry`	`RetryCallback \| None`	Called with a `RetryState` before each wait. May be async for async functions.	`None`
`delay_from`	`Callable[[BaseException], float \| None] \| None`	Called with the exception; a number is the exact wait for this attempt (no backoff, no jitter; still capped by `max_delay` and `max_time`), `None` means use the normal backoff. See `retry_after_header`.	`None`
`budget`	`RetryBudget \| None`	A `RetryBudget` shared with other retried code. When it refuses a retry the last exception is re-raised (or the last result returned) at once.	`_UNSET`
`logger`	`LoggerLike`	Logger (or logger name) for the per-retry message.	`None`
`log_level`	`int \| None`	Level of the per-retry message, or `None` to not log.	`_UNSET`
`log_exc_info`	`bool`	Attach the traceback to the per-retry log record.	`False`

Returns:

Type	Description
`Any`	The wrapped function, or a decorator when `func` is omitted.

Raises:

Type	Description
`ValueError`	If an option is out of range (raised when decorating).
`TypeError`	If `budget` is not a `RetryBudget`, or a sync function is given an async `retry_if`, `retry_if_result` or `on_retry` (raised when decorating).

Example

@retry(max_attempts=5, delay=0.5, backoff=2, on=ConnectionError) ... def fetch() -> str: ...

retrying ¶

retrying(*, max_attempts: int | None = _UNSET, delay: float = _UNSET, backoff: float = _UNSET, max_delay: float | None = _UNSET, jitter: float = _UNSET, max_time: float | None = _UNSET, on: ExceptionTypes = Exception, retry_if: ExceptionPredicate | None = None, retry_if_result: ResultPredicate | None = None, on_retry: RetryCallback | None = None, delay_from: Callable[[BaseException], float | None] | None = None, budget: RetryBudget | None = _UNSET, logger: LoggerLike = None, log_level: int | None = _UNSET, log_exc_info: bool = False, name: str = 'retrying') -> Retrying

Retry an inline block of code instead of a whole function.

Takes the same options as retry and behaves identically (backoff, jitter, max_time, delay_from, budget, cancellation rules, events, test overrides). Put the risky code in with attempt:: a retryable exception is swallowed and the loop waits and continues; success ends the loop; the final failure propagates.

Use for in sync code and async for in async code. Do not use for inside a coroutine, it would block the event loop while waiting. Async retry_if, retry_if_result or on_retry callables need async for and async with attempt:; with plain with attempt: they raise TypeError.

For retry_if_result, assign the block's value to attempt.result; a rejected value is retried, and when no retry is left the loop ends normally with the last value (like retry, nothing is raised).

Parameters:

Name	Type	Description	Default
`max_attempts`	`int \| None`	Total attempts allowed, or `None` for no limit.	`_UNSET`
`delay`	`float`	Base wait in seconds before the second attempt.	`_UNSET`
`backoff`	`float`	Multiplier applied to the wait after each failure (>= 1).	`_UNSET`
`max_delay`	`float \| None`	Upper bound for the backoff wait (before jitter).	`_UNSET`
`jitter`	`float`	Up to this many random extra seconds are added to each wait.	`_UNSET`
`max_time`	`float \| None`	Total time budget in seconds from the first attempt.	`_UNSET`
`on`	`ExceptionTypes`	Exception type(s) that trigger a retry.	`Exception`
`retry_if`	`ExceptionPredicate \| None`	Extra predicate on the exception; retry only if it returns True.	`None`
`retry_if_result`	`ResultPredicate \| None`	Predicate on `attempt.result`; True means "retry".	`None`
`on_retry`	`RetryCallback \| None`	Called with a `RetryState` before each wait.	`None`
`delay_from`	`Callable[[BaseException], float \| None] \| None`	Exact wait derived from the exception, or `None` for backoff.	`None`
`budget`	`RetryBudget \| None`	A shared `RetryBudget`; when it refuses, the loop stops retrying.	`_UNSET`
`logger`	`LoggerLike`	Logger (or logger name) for the per-retry message.	`None`
`log_level`	`int \| None`	Level of the per-retry message, or `None` to not log.	`_UNSET`
`log_exc_info`	`bool`	Attach the traceback to the per-retry log record.	`False`
`name`	`str`	Name used in log messages and events.	`'retrying'`

Returns:

Type	Description
`Retrying`	A `Retrying` object to iterate over.

Raises:

Type	Description
`ValueError`	If an option is out of range.
`TypeError`	If `budget` is not a `RetryBudget`.

Example

for attempt in retrying(max_attempts=3, delay=0.1, on=ConnectionError): ... with attempt: ... do_work() # doctest: +SKIP

retry_on_exception ¶

retry_on_exception(*exceptions: type[BaseException], max_attempts: int | None = 3, delay: float = 0.0, backoff: float = 1.0, max_delay: float | None = None, jitter: float = 0.0, retry_if: Callable[[BaseException], bool] | None = None, on_retry: Callable[[RetryState], None] | None = None, logger: LoggerLike = None) -> Callable[[F], F]

Deprecated alias for retry(on=exceptions or Exception, ...).

Parameters:

Name	Type	Description	Default
`*exceptions`	`type[BaseException]`	Exception types to retry on (default `Exception`).	`()`
`max_attempts`	`int \| None`	Total calls allowed, or `None` for no limit.	`3`
`delay`	`float`	Base wait in seconds between attempts.	`0.0`
`backoff`	`float`	Multiplier applied to the wait after each failure.	`1.0`
`max_delay`	`float \| None`	Upper bound for the wait.	`None`
`jitter`	`float`	Random extra seconds added to each wait.	`0.0`
`retry_if`	`Callable[[BaseException], bool] \| None`	Extra predicate deciding whether to retry.	`None`
`on_retry`	`Callable[[RetryState], None] \| None`	Callback invoked before each wait.	`None`
`logger`	`LoggerLike`	Logger (or name) for retry messages.	`None`

Returns:

Type	Description
`Callable[[F], F]`	A decorator. Applying it emits a `DeprecationWarning`.

Raises:

Type	Description
`ValueError`	If an option is out of range.

Example

@retry_on_exception(ValueError) # use retry(on=ValueError) instead ... def parse() -> None: ...

fallback ¶

fallback(value: T, /, *, on: ExceptionTypes = ..., logger: LoggerLike = ...) -> _Decorator[T]

fallback(*, handler: Callable[[BaseException], T], on: ExceptionTypes = ..., logger: LoggerLike = ...) -> _Decorator[T]

fallback(value: Any = _MISSING, /, *, handler: Callable[[BaseException], Any] | None = None, on: ExceptionTypes = Exception, logger: LoggerLike = None) -> Any

Return a default value (or a handler's result) when the function raises.

Parameters:

Name	Type	Description	Default
`value`	`Any`	The value to return on a matching exception.	`_MISSING`
`handler`	`Callable[[BaseException], Any] \| None`	Alternatively, a callable `handler(exc)` whose result is returned.	`None`
`on`	`ExceptionTypes`	Exception type(s) that trigger the fallback.	`Exception`
`logger`	`LoggerLike`	Logger (or name) for the WARNING message (logged with traceback).	`None`

Returns:

Type	Description
`Any`	A decorator for sync or async functions. For async functions the handler is
`Any`	still a regular (sync) function.

Raises:

Type	Description
`ValueError`	If neither or both of `value` and `handler` are given.

Example

@fallback(0, on=KeyError) ... def lookup(key: str) -> int: ...

suppress ¶

suppress(exc: type[BaseException], /, *more: type[BaseException], logger: LoggerLike = ...) -> _Decorator[None]

suppress(func: Callable[P, Coroutine[Any, Any, R]]) -> Callable[P, Coroutine[Any, Any, R | None]]

suppress(func: Callable[P, R]) -> Callable[P, R | None]

suppress(*, logger: LoggerLike = ...) -> _Decorator[None]

suppress(*exceptions: Any, logger: LoggerLike = None) -> Any

Swallow matching exceptions; the call then returns None.

Usable bare (@suppress, which suppresses Exception) or with exception types (@suppress(KeyError)). Suppressed errors are logged at DEBUG.

Parameters:

Name	Type	Description	Default
`*exceptions`	`Any`	Exception types to swallow (default `Exception`).	`()`
`logger`	`LoggerLike`	Logger (or name) for the DEBUG message.	`None`

Returns:

Type	Description
`Any`	The wrapped function, or a decorator when exception types are given.

Example

@suppress(KeyError) ... def first(d: dict[str, int]) -> int: ...

Retry and fallback

retry ¶

RetryState dataclass ¶

RetryStats ¶

snapshot ¶

reset ¶

RetryBudgetStats dataclass ¶

RetryBudget ¶

stats property ¶

deposit ¶

try_withdraw ¶

reset ¶

Attempt ¶

result property writable ¶

Retrying ¶

stats property ¶

retry_stats ¶

retry_after_header ¶

retry ¶

retrying ¶

retry_on_exception ¶

fallback ¶

suppress ¶

RetryState `dataclass` ¶

RetryBudgetStats `dataclass` ¶

stats `property` ¶

result `property` `writable` ¶

stats `property` ¶