API Reference¶
- stamina.retry(*, on, attempts=10, timeout=45.0, wait_initial=0.1, wait_max=5.0, wait_jitter=1.0, wait_exp_base=2.0)[source]¶
Retry if one of configured exceptions are raised.
The backoff delays between retries grow exponentially plus a random jitter.
The backoff for retry attempt number attempt is computed as:
\[min(wait\_max, wait\_initial * wait\_exp\_base^{attempt - 1} + random(0, wait\_jitter))\]Since \(x^0\) is always 1, the first backoff is within the interval \([wait\_initial,wait\_initial+wait\_jitter]\). Thus, with default values between 0.1 and 1.1 seconds.
If all retries fail, the last exception is let through.
All float-based time parameters are in seconds.
- Parameters:
on (Type[Exception] | Tuple[Type[Exception], ...] | Callable[[Exception], bool]) –
An Exception or a tuple of Exceptions on which the decorated callable will be retried.
You can also pass a predicate in the form of a callable that takes an exception and returns a bool which decides whether the exception should be retried – True meaning yes.
This allows more fine-grained control over when to retry. For example, to only retry on HTTP errors in the 500s range that indicate server errors, but not those in the 400s which indicate a client error.
There is no default – you must pass this explicitly.
attempts (int | None) – Maximum total number of attempts. Can be combined with timeout.
timeout (float | timedelta | None) – Maximum total time for all retries. Can be combined with attempts.
wait_initial (float | timedelta) – Minimum backoff before the first retry.
wait_max (float | timedelta) – Maximum backoff time between retries at any time.
wait_jitter (float | timedelta) – Maximum jitter that is added to retry back-off delays (the actual jitter added is a random number between 0 and wait_jitter)
wait_exp_base (float) – The exponential base used to compute the retry backoff.
Changed in version 23.1.0: All time-related parameters can now be specified as a
datetime.timedelta
.Added in version 23.3.0: Trio support.
Added in version 24.3.0: on can be a callable now.
- stamina.retry_context(on, attempts=10, timeout=45.0, wait_initial=0.1, wait_max=5.0, wait_jitter=1.0, wait_exp_base=2.0)[source]¶
Iterator that yields context managers that can be used to retry code blocks.
Arguments have the same meaning as for
stamina.retry()
.Added in version 23.1.0.
Added in version 23.3.0: Trio support.
- class stamina.Attempt(attempt, next_wait_fn)[source]¶
A context manager that can be used to retry code blocks.
Instances are yielded by the
stamina.retry_context()
iterator.Added in version 23.2.0.
- class stamina.RetryingCaller(attempts=10, timeout=45.0, wait_initial=0.1, wait_max=5.0, wait_jitter=1.0, wait_exp_base=2.0)[source]¶
Call your callables with retries.
Arguments have the same meaning as for
stamina.retry()
.Tip
Instances of
RetryingCaller
may be reused because they internally create a newretry_context()
iterator on each call.Added in version 24.2.0.
For example:
def do_something_with_url(url, some_kw): resp = httpx.get(url).raise_for_status() ... rc = stamina.RetryingCaller(attempts=5) rc(httpx.HTTPError, do_something_with_url, f"https://httpbin.org/status/404", some_kw=42) # Equivalent: bound_rc = rc.on(httpx.HTTPError) bound_rc(do_something_with_url, f"https://httpbin.org/status/404", some_kw=42)
Both calls to
rc
andbound_rc
rundo_something_with_url(f"https://httpbin.org/status/404", some_kw=42)
and retry on
httpx.HTTPError
.- __call__(on, callable_, /, *args, **kwargs)[source]¶
Call
callable_(*args, **kw)
with retries if on is raised.
- on(on, /)[source]¶
Create a new instance of
BoundRetryingCaller
with the same parameters, but bound to a specific exception type.Added in version 24.2.0.
- class stamina.BoundRetryingCaller(caller, on)[source]¶
Same as
RetryingCaller
, but pre-bound to a specific exception type.Caution
Returned by
RetryingCaller.on()
– do not instantiate directly.Added in version 24.2.0.
- __call__(callable_, /, *args, **kwargs)[source]¶
Same as
RetryingCaller.__call__()
, except retry on the exception that is bound to this instance.
- class stamina.AsyncRetryingCaller(attempts=10, timeout=45.0, wait_initial=0.1, wait_max=5.0, wait_jitter=1.0, wait_exp_base=2.0)[source]¶
Same as
RetryingCaller
, but for async callables.Added in version 24.2.0.
- async __call__(on, callable_, /, *args, **kwargs)[source]¶
Same as
RetryingCaller.__call__()
, but callable_ is awaited.
- on(on, /)[source]¶
Create a new instance of
BoundAsyncRetryingCaller
with the same parameters, but bound to a specific exception type.Added in version 24.2.0.
- class stamina.BoundAsyncRetryingCaller(caller, on)[source]¶
Same as
BoundRetryingCaller
, but for async callables.Caution
Returned by
AsyncRetryingCaller.on()
– do not instantiate directly.Added in version 24.2.0.
- async __call__(callable_, /, *args, **kwargs)[source]¶
Same as
AsyncRetryingCaller.__call__()
, except retry on the exception that is bound to this instance.
Activation and deactivation¶
Testing¶
See also
- stamina.set_testing(testing, *, attempts=1, cap=False)[source]¶
Activate or deactivate test mode.
In testing mode, backoffs are disabled, and attempts are set to attempts.
If cap is True, the number of attempts is not set but capped at attempts. This means that if attempts is greater than the number of attempts specified by the user, the user’s value is used.
Is idempotent and can be called repeatedly with the same values.
Added in version 24.3.0.
Added in version 25.1.0: cap
Instrumentation¶
- stamina.instrumentation.set_on_retry_hooks(hooks)[source]¶
Set hooks that are called after a retry has been scheduled.
- Parameters:
hooks (Iterable[RetryHook | RetryHookFactory] | None) – Hooks to call after a retry has been scheduled. Passing None resets to default. To deactivate instrumentation, pass an empty iterable.
Added in version 23.2.0.
- stamina.instrumentation.get_on_retry_hooks()[source]¶
Get hooks that are called after a retry has been scheduled.
- Returns:
Hooks that will run if a retry is scheduled. Factories are called if they haven’t already.
- Return type:
Added in version 23.2.0.
- class stamina.instrumentation.RetryHook[source]¶
A callable that gets called after an attempt has failed and a retry has been scheduled.
This is a
typing.Protocol
that can be implemented by any callable that takes one argument of typeRetryDetails
and returns None.Added in version 23.2.0.
For example:
def print_hook(details: stamina.instrumentation.RetryDetails) -> None: print("a retry has been scheduled!", details) stamina.set_on_retry_hooks([print_hook])
- class stamina.instrumentation.RetryHookFactory(hook_factory)[source]¶
Wraps a callable that returns a
RetryHook
.They are called on the first scheduled retry and can be used to delay initialization. If you need to pass arguments, you can do that using
functools.partial()
.Added in version 23.2.0.
For example, if your instrumentation needs to import a module
something_expensive
which takes a long time to import, you can delay it until the first retry (or call tostamina.instrumentation.get_on_retry_hooks()
):from stamina.instrumentation import RetryHookFactory, RetryDetails def init_with_expensive_import(): import something_expensive def do_something(details: RetryDetails) -> None: something_expensive.do_something(details) return do_something stamina.set_on_retry_hooks([RetryHookFactory(init_with_expensive_import)])
- class stamina.instrumentation.RetryDetails(name, args, kwargs, retry_num, wait_for, waited_so_far, caused_by)[source]¶
Details about a retry attempt that are passed into
RetryHook
s.Added in version 23.2.0.
Integrations¶
- stamina.instrumentation.StructlogOnRetryHook¶
Pass this object to
stamina.instrumentation.set_on_retry_hooks()
to activate structlog integration.Is active by default if structlog can be imported.
See also
Added in version 23.2.0.
- stamina.instrumentation.LoggingOnRetryHook¶
Pass this object to
stamina.instrumentation.set_on_retry_hooks()
to activatelogging
integration.Is active by default if structlog can not be imported.
See also
Added in version 23.2.0.
- stamina.instrumentation.PrometheusOnRetryHook¶
Pass this object to
stamina.instrumentation.set_on_retry_hooks()
to activate Prometheus integration.Is active by default if prometheus-client can be imported.
See also
Added in version 23.2.0.