API Reference

Retry Main API

tenacity.retry(func: WrappedFn) → WrappedFn

tenacity.retry(sleep: Callable[[Union[int, float]], 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: Optional[Callable[[RetryCallState], None]] = None, reraise: bool = False, retry_error_cls: Type[RetryError] = RetryError, retry_error_callback: Optional[Callable[[RetryCallState], Any]] = 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[[~typing.Union[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.Optional[~typing.Callable[[RetryCallState], None]] = None, reraise: bool = False, retry_error_cls: ~typing.Type[~tenacity.RetryError] = <class 'tenacity.RetryError'>, retry_error_callback: ~typing.Optional[~typing.Callable[[RetryCallState], ~typing.Any]] = None)

Retrying controller.

class tenacity.AsyncRetrying(sleep: ~typing.Callable[[float], ~typing.Awaitable[~typing.Any]] = <function 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)

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.Union[~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: Optional[str] = None, match: Optional[str] = None)

Retries if an exception message equals or matches.

class tenacity.retry.retry_if_exception_type(exception_types: ~typing.Union[~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: Optional[str] = None, match: Optional[str] = None)

Retries until an exception message equals or matches.

class tenacity.retry.retry_if_not_exception_type(exception_types: ~typing.Union[~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.Union[~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: Union[int, float, timedelta])

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

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_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: Union[int, float] = 1, max: Union[int, float, timedelta] = 4.611686018427388e+18, exp_base: Union[int, float] = 2, min: Union[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: Union[int, float, timedelta])

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

class tenacity.wait.wait_incrementing(start: Union[int, float, timedelta] = 0, increment: Union[int, float, timedelta] = 100, max: Union[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: Union[int, float, timedelta] = 0, max: Union[int, float, timedelta] = 1)

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

class tenacity.wait.wait_random_exponential(multiplier: Union[int, float] = 1, max: Union[int, float, timedelta] = 4.611686018427388e+18, exp_base: Union[int, float] = 2, min: Union[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.