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)[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.

Generators are restarted from the beginning on each retry.

Note

Being able to asend() and athrow() into wrapped async generators introduced nontrivial complexity in the implementation and is therefore provisional. If supporting these features causes problems, they may be removed again in a future version.

Warning

It is possible to get unbounded retries by passing None for attempts (= infinite attempts) and timeout (= no timeout).

Parameters:

on (type[Exception] | tuple[type[Exception], ...] | Callable[[Exception], bool | float | timedelta]) –
An Exception or a tuple of Exceptions on which the decorated callable will be retried.

You can also pass a backoff hook 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.

For even more control, the hook may return a float or a datetime.timedelta to specify a custom backoff that overrides the default backoff. This is useful when the error carries information like a Retry-After header. A custom backoff is not part of the exponential backoff machinery so none of the other backoff parameters apply to it.

There is no default for the on parameter – you must pass on explicitly.
attempts (int | None) – Maximum total number of attempts. Can be combined with timeout. None means no limit to attempts.
timeout (float | timedelta | None) – Maximum total time for all retries. Can be combined with attempts. None means no timeout.
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.

Added in version 25.2.0: Generator functions and async generator functions are now retried, too.

Added in version 25.2.0: An on backoff hook can now return a float or a datetime.timedelta to specify a custom backoff that overrides the default backoff.

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)[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.

property next_wait: float¶: The number of seconds of backoff before the next attempt if this attempt fails.

Warning

This value does not include a possible random jitter and is therefore just a lower bound of the actual value.

Added in version 24.3.0.

Changed in version 25.2.0: Fixed off-by-one error when passing the attempt number to the next_wait_fn.

property num: int¶: The number of the current attempt.

class stamina.RetryingCaller(attempts=10, timeout=45.0, wait_initial=0.1, wait_max=5.0, wait_jitter=1.0, wait_exp_base=2)[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 new retry_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 and bound_rc run

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

Parameters:

on (type[Exception] | tuple[type[Exception], ...] | Callable[[Exception], bool | float | timedelta]) – Exception(s) to retry on.
callable_ (Callable[[~P], T]) – Callable to call.
args (~P) – Positional arguments to pass to callable_.
kw – Keyword arguments to pass to callable_.

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)[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¶

stamina.set_active(active)[source]¶

Activate or deactivate retrying.

Is idempotent and can be called repeatedly with the same value.

stamina.is_active()[source]¶

Check whether retrying is active.

Returns:: Whether retrying is active.
Return type:: bool

Testing¶

See also

Testing

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

Added in version 25.1.0: Can be used as a context manager.

stamina.is_testing()[source]¶

Check whether test mode is enabled.

Added in version 24.3.0.

Instrumentation¶

See also

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:: tuple[RetryHook, …]

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 type RetryDetails and returns None.

If the hook returns a context manager, it will be entered when the retry is scheduled and exited right before the retry is attempted.

Added in version 23.2.0.

Added in version 25.1.0: Added support for context managers.

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 to stamina.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 RetryHooks.

name¶

Name of the callable that is being retried.

Type:: str

args¶

Positional arguments that were passed to the callable.

Type:: tuple[object, …]

kwargs¶

Keyword arguments that were passed to the callable.

Type:: dict[str, object]

retry_num¶

Number of the retry attempt. Starts at 1 after the first failure.

Type:: int

wait_for¶

Time in seconds that stamina will wait before the next attempt.

Type:: float

waited_so_far¶

Time in seconds that stamina has waited so far for the current callable.

Type:: float

caused_by¶

Exception that caused the retry attempt.

Type:: Exception

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

structlog

Added in version 23.2.0.

stamina.instrumentation.LoggingOnRetryHook¶

Pass this object to stamina.instrumentation.set_on_retry_hooks() to activate logging integration.

Is active by default if structlog can not be imported.

See also

Standard library’s logging

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

Prometheus

Added in version 23.2.0.

stamina.instrumentation.get_prometheus_counter()[source]¶

Return the Prometheus counter for the number of retries.

Returns:: If active, the Prometheus counter for the number of retries. None otherwise.
Return type:: Counter | None

Added in version 23.2.0.