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.

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.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 new retry_context() iterator on each call.

New 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], ...]) – 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.

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

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

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

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

New 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

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

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 type RetryDetails 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 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

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.