API Reference

Retry Main API

tenacity.retry(func: WrappedFn) → WrappedFn

tenacity.retry(sleep: Callable[[int | float], Awaitable[None] | None] = sleep, stop: StopBaseT = stop_never, wait: WaitBaseT = wait_none(), retry: RetryBaseT = retry_if_exception_type(), before: Callable[[RetryCallState], None] = before_nothing, after: Callable[[RetryCallState], None] = after_nothing, before_sleep: Callable[[RetryCallState], None] | None = None, reraise: bool = False, retry_error_cls: Type[RetryError] = RetryError, retry_error_callback: Callable[[RetryCallState], Any] | None = None) → Callable[[WrappedFn], WrappedFn]

Wrap a function with a new Retrying object.

Parameters:

• dargs – positional arguments passed to Retrying object

• dkw – keyword arguments passed to the Retrying object

class tenacity.Retrying(sleep: ~typing.Callable[[int | float], None] = <function sleep>, stop: StopBaseT = <tenacity.stop._stop_never object>, wait: WaitBaseT = <tenacity.wait.wait_none object>, retry: RetryBaseT = <tenacity.retry.retry_if_exception_type object>, before: ~typing.Callable[[RetryCallState], None] = <function before_nothing>, after: ~typing.Callable[[RetryCallState], None] = <function after_nothing>, before_sleep: ~typing.Callable[[RetryCallState], None] | None = None, reraise: bool = False, retry_error_cls: ~typing.Type[~tenacity.RetryError] = <class 'tenacity.RetryError'>, retry_error_callback: ~typing.Callable[[RetryCallState], ~typing.Any] | None = None)

Retrying controller.

class tenacity.AsyncRetrying(sleep: ~typing.Callable[[float], ~typing.Awaitable[~typing.Any]] = <function asyncio_sleep>, **kwargs: ~typing.Any)

wraps(fn: WrappedFn) → WrappedFn

Wrap a function for retrying.

Parameters:

f – A function to wraps for retrying.

class tenacity.tornadoweb.TornadoRetrying(sleep: typing.Callable[[float], Future[None]] = <function sleep>, **kwargs: ~typing.Any)

class tenacity.RetryCallState(retry_object: BaseRetrying, fn: WrappedFn | None, args: Any, kwargs: Any)

State related to a single call wrapped with Retrying.

args

Arguments of the function wrapped by this retry call

attempt_number: int

The number of the current attempt

fn

Function wrapped by this retry call

idle_for: float

Time spent sleeping in retries

kwargs

Keyword arguments of the function wrapped by this retry call

next_action: RetryAction | None

Next action as decided by the retry manager

outcome: Future | None

Last outcome (result or exception) produced by the function

outcome_timestamp: float | None

Timestamp of the last outcome

retry_object

Retry manager object

start_time

Retry call start timestamp

upcoming_sleep: float

Next sleep time as decided by the retry manager.

After Functions

Those functions can be used as the after keyword argument of tenacity.retry().

tenacity.after.after_log(logger: logging.Logger, log_level: int, sec_format: str = '%0.3f') → Callable[[RetryCallState], None]

After call strategy that logs to some logger the finished attempt.

tenacity.after.after_nothing(retry_state: RetryCallState) → None

After call strategy that does nothing.

Before Functions

Those functions can be used as the before keyword argument of tenacity.retry().

tenacity.before.before_log(logger: logging.Logger, log_level: int) → Callable[[RetryCallState], None]

Before call strategy that logs to some logger the attempt.

tenacity.before.before_nothing(retry_state: RetryCallState) → None

Before call strategy that does nothing.

Before Sleep Functions

Those functions can be used as the before_sleep keyword argument of tenacity.retry().

tenacity.before_sleep.before_sleep_log(logger: logging.Logger, log_level: int, exc_info: bool = False) → Callable[[RetryCallState], None]

Before call strategy that logs to some logger the attempt.

tenacity.before_sleep.before_sleep_nothing(retry_state: RetryCallState) → None

Before call strategy that does nothing.

Nap Functions

Those functions can be used as the sleep keyword argument of tenacity.retry().

tenacity.nap.sleep(seconds: float) → None

Sleep strategy that delays execution for a given number of seconds.

This is the default strategy, and may be mocked out for unit testing.

class tenacity.nap.sleep_using_event(event: threading.Event)

Sleep strategy that waits on an event to be set.

Retry Functions

Those functions can be used as the retry keyword argument of tenacity.retry().

class tenacity.retry.retry_all(*retries: retry_base)

Retries if all the retries condition are valid.

class tenacity.retry.retry_any(*retries: retry_base)

Retries if any of the retries condition is valid.

class tenacity.retry.retry_base

Abstract base class for retry strategies.

class tenacity.retry.retry_if_exception(predicate: Callable[[BaseException], bool])

Retry strategy that retries if an exception verifies a predicate.

class tenacity.retry.retry_if_exception_cause_type(exception_types: ~typing.Type[BaseException] | ~typing.Tuple[~typing.Type[BaseException], ...] = <class 'Exception'>)

Retries if any of the causes of the raised exception is of one or more types.

The check on the type of the cause of the exception is done recursively (until finding an exception in the chain that has no __cause__)

class tenacity.retry.retry_if_exception_message(message: str | None = None, match: str | None = None)

Retries if an exception message equals or matches.

class tenacity.retry.retry_if_exception_type(exception_types: ~typing.Type[BaseException] | ~typing.Tuple[~typing.Type[BaseException], ...] = <class 'Exception'>)

Retries if an exception has been raised of one or more types.

class tenacity.retry.retry_if_not_exception_message(message: str | None = None, match: str | None = None)

Retries until an exception message equals or matches.

class tenacity.retry.retry_if_not_exception_type(exception_types: ~typing.Type[BaseException] | ~typing.Tuple[~typing.Type[BaseException], ...] = <class 'Exception'>)

Retries except an exception has been raised of one or more types.

class tenacity.retry.retry_if_not_result(predicate: Callable[[Any], bool])

Retries if the result refutes a predicate.

class tenacity.retry.retry_if_result(predicate: Callable[[Any], bool])

Retries if the result verifies a predicate.

class tenacity.retry.retry_unless_exception_type(exception_types: ~typing.Type[BaseException] | ~typing.Tuple[~typing.Type[BaseException], ...] = <class 'Exception'>)

Retries until an exception is raised of one or more types.

Stop Functions

Those functions can be used as the stop keyword argument of tenacity.retry().

class tenacity.stop.stop_after_attempt(max_attempt_number: int)

Stop when the previous attempt >= max_attempt.

class tenacity.stop.stop_after_delay(max_delay: int | float | timedelta)

Stop when the time from the first attempt >= limit.

Note: max_delay will be exceeded, so when used with a wait, the actual total delay will be greater than max_delay by some of the final sleep period before max_delay is exceeded.

If you need stricter timing with waits, consider stop_before_delay instead.

class tenacity.stop.stop_all(*stops: stop_base)

Stop if all the stop conditions are valid.

class tenacity.stop.stop_any(*stops: stop_base)

Stop if any of the stop condition is valid.

class tenacity.stop.stop_base

Abstract base class for stop strategies.

class tenacity.stop.stop_before_delay(max_delay: int | float | timedelta)

Stop right before the next attempt would take place after the time from the first attempt >= limit.

Most useful when you are using with a wait function like wait_random_exponential, but need to make sure that the max_delay is not exceeded.

class tenacity.stop.stop_when_event_set(event: threading.Event)

Stop when the given event is set.

Wait Functions

Those functions can be used as the wait keyword argument of tenacity.retry().

class tenacity.wait.wait_base

Abstract base class for wait strategies.

class tenacity.wait.wait_chain(*strategies: wait_base)

Chain two or more waiting strategies.

If all strategies are exhausted, the very last strategy is used thereafter.

For example:

@retry(wait=wait_chain(*[wait_fixed(1) for i in range(3)] +
                       [wait_fixed(2) for j in range(5)] +
                       [wait_fixed(5) for k in range(4)))
def wait_chained():
    print("Wait 1s for 3 attempts, 2s for 5 attempts and 5s
           thereafter.")

class tenacity.wait.wait_combine(*strategies: wait_base)

Combine several waiting strategies.

class tenacity.wait.wait_exponential(multiplier: int | float = 1, max: int | float | timedelta = 4.611686018427388e+18, exp_base: int | float = 2, min: int | float | timedelta = 0)

Wait strategy that applies exponential backoff.

It allows for a customized multiplier and an ability to restrict the upper and lower limits to some maximum and minimum value.

The intervals are fixed (i.e. there is no jitter), so this strategy is suitable for balancing retries against latency when a required resource is unavailable for an unknown duration, but not suitable for resolving contention between multiple processes for a shared resource. Use wait_random_exponential for the latter case.

class tenacity.wait.wait_exponential_jitter(initial: float = 1, max: float = 4.611686018427388e+18, exp_base: float = 2, jitter: float = 1)

Wait strategy that applies exponential backoff and jitter.

It allows for a customized initial wait, maximum wait and jitter.

This implements the strategy described here: https://cloud.google.com/storage/docs/retry-strategy

The wait time is min(initial * 2**n + random.uniform(0, jitter), maximum) where n is the retry count.

class tenacity.wait.wait_fixed(wait: int | float | timedelta)

Wait strategy that waits a fixed amount of time between each retry.

class tenacity.wait.wait_incrementing(start: int | float | timedelta = 0, increment: int | float | timedelta = 100, max: int | float | timedelta = 4.611686018427388e+18)

Wait an incremental amount of time after each attempt.

Starting at a starting value and incrementing by a value for each attempt (and restricting the upper limit to some maximum value).

class tenacity.wait.wait_none

Wait strategy that doesn’t wait at all before retrying.

class tenacity.wait.wait_random(min: int | float | timedelta = 0, max: int | float | timedelta = 1)

Wait strategy that waits a random amount of time between min/max.

class tenacity.wait.wait_random_exponential(multiplier: int | float = 1, max: int | float | timedelta = 4.611686018427388e+18, exp_base: int | float = 2, min: int | float | timedelta = 0)

Random wait with exponentially widening window.

An exponential backoff strategy used to mediate contention between multiple uncoordinated processes for a shared resource in distributed systems. This is the sense in which “exponential backoff” is meant in e.g. Ethernet networking, and corresponds to the “Full Jitter” algorithm described in this blog post:

https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/

Each retry occurs at a random time in a geometrically expanding interval. It allows for a custom multiplier and an ability to restrict the upper limit of the random interval to some maximum value.

Example:

wait_random_exponential(multiplier=0.5,  # initial window 0.5s
                        max=60)          # max 60s timeout

When waiting for an unavailable resource to become available again, as opposed to trying to resolve contention for a shared resource, the wait_exponential strategy (which uses a fixed interval) may be preferable.