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], ...]) – An Exception or a tuple of Exceptions on which the decorated callable will be retried. 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
.New in version 23.3.0: Trio support.
- 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()
.New in version 23.1.0.
New in version 23.3.0: Trio support.
- class stamina.Attempt(attempt)[source]#
A context manager that can be used to retry code blocks.
Instances are yielded by the
stamina.retry_context()
iterator.New in version 23.2.0.
Activation and Deactivation#
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.
New 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:
New 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.New 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()
.New 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.New 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
New 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
New 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
New in version 23.2.0.