tennacity 1.2.0

A module that prints Hello World on import

Tenacity

Reliable and configurable retry behavior for Python.

This is an independently written README.md for the Tenacity Python package.
It is not the upstream project's official README.

Tenacity helps applications recover from temporary failures by retrying functions or code blocks according to configurable policies.

Typical uses include:

• Retrying network requests after timeouts
• Reconnecting to temporarily unavailable services
• Repeating database operations after transient errors
• Polling until a result becomes available
• Applying retry behavior to synchronous and asynchronous functions

Features

• Decorator-based retry configuration
• Attempt-based and time-based stopping rules
• Fixed, random, exponential, and jittered wait strategies
• Retry rules based on exceptions or return values
• Synchronous and asynchronous support
• Retryable code blocks with Retrying and AsyncRetrying
• Logging and custom callbacks
• Runtime policy overrides
• Retry statistics

Requirements

This README targets modern Tenacity releases and Python 3.10 or newer.

Installation

Install Tenacity from PyPI:

python -m pip install tenacity

Verify the installation:

python -c "import tenacity; print('Tenacity is installed')"

Quick Start

from tenacity import retry, stop_after_attempt, wait_fixed


class TemporaryServiceError(Exception):
    pass


attempts = 0


@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    reraise=True,
)
def load_data() -> str:
    global attempts
    attempts += 1

    if attempts < 3:
        raise TemporaryServiceError("Service is temporarily unavailable")

    return "Data loaded"


print(load_data())

The function is attempted up to three times, with a one-second delay between failed attempts.

Important Default Behavior

Using @retry without arguments retries indefinitely and does not wait between attempts:

from tenacity import retry


@retry
def unstable_operation() -> None:
    raise RuntimeError("This will be retried forever")

Production code should normally define a stopping rule and a waiting strategy.

Stop After a Number of Attempts

from tenacity import retry, stop_after_attempt


@retry(
    stop=stop_after_attempt(5),
    reraise=True,
)
def connect() -> None:
    raise ConnectionError("Connection failed")

The first call counts as attempt 1, so this configuration allows no more than five total calls.

Stop After a Time Limit

from tenacity import retry, stop_after_delay, wait_fixed


@retry(
    stop=stop_after_delay(15),
    wait=wait_fixed(2),
    reraise=True,
)
def wait_for_service() -> None:
    raise ConnectionError("Service is not ready")

Combine Stop Conditions

Use | to stop when either condition becomes true:

from tenacity import retry, stop_after_attempt, stop_after_delay


@retry(
    stop=stop_after_attempt(6) | stop_after_delay(20),
    reraise=True,
)
def perform_operation() -> None:
    raise RuntimeError("Temporary failure")

This policy stops after six attempts or twenty seconds, whichever happens first.

Wait Strategies

Fixed Wait

from tenacity import retry, stop_after_attempt, wait_fixed


@retry(
    stop=stop_after_attempt(4),
    wait=wait_fixed(2),
    reraise=True,
)
def fixed_wait_operation() -> None:
    raise TimeoutError("Timed out")

Random Wait

from tenacity import retry, stop_after_attempt, wait_random


@retry(
    stop=stop_after_attempt(4),
    wait=wait_random(min=1, max=3),
    reraise=True,
)
def random_wait_operation() -> None:
    raise TimeoutError("Timed out")

Exponential Backoff

from tenacity import retry, stop_after_attempt, wait_exponential


@retry(
    stop=stop_after_attempt(6),
    wait=wait_exponential(
        multiplier=1,
        min=1,
        max=30,
    ),
    reraise=True,
)
def backoff_operation() -> None:
    raise ConnectionError("Remote service unavailable")

Exponential Backoff with Jitter

Randomized exponential backoff is useful when many clients may retry the same service simultaneously.

from tenacity import retry, stop_after_attempt, wait_random_exponential


@retry(
    stop=stop_after_attempt(6),
    wait=wait_random_exponential(
        multiplier=1,
        min=1,
        max=30,
    ),
    reraise=True,
)
def distributed_operation() -> None:
    raise ConnectionError("Remote service unavailable")

Combine Wait Strategies

Wait strategies can be combined with +:

from tenacity import retry, stop_after_attempt, wait_fixed, wait_random


@retry(
    stop=stop_after_attempt(5),
    wait=wait_fixed(2) + wait_random(0, 1),
    reraise=True,
)
def combined_wait_operation() -> None:
    raise RuntimeError("Temporary failure")

This waits at least two seconds and adds up to one second of random delay.

Retry Selected Exceptions

Retry only failures that are likely to be temporary.

from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_random_exponential,
)


@retry(
    retry=retry_if_exception_type(
        (TimeoutError, ConnectionError)
    ),
    stop=stop_after_attempt(5),
    wait=wait_random_exponential(
        multiplier=0.5,
        max=10,
    ),
    reraise=True,
)
def request_remote_data() -> bytes:
    raise TimeoutError("The request timed out")

Exceptions not included in the retry rule are raised immediately.

Retry Based on a Return Value

from typing import Optional

from tenacity import (
    retry,
    retry_if_result,
    stop_after_attempt,
    wait_fixed,
)


def result_is_missing(value: Optional[str]) -> bool:
    return value is None


@retry(
    retry=retry_if_result(result_is_missing),
    stop=stop_after_attempt(5),
    wait=wait_fixed(1),
)
def find_job_result() -> Optional[str]:
    return None

The function is retried whenever its return value satisfies the predicate.

Raise the Original Exception

By default, exhausting the retry policy raises RetryError. Use reraise=True to raise the last underlying exception instead:

from tenacity import retry, stop_after_attempt


@retry(
    stop=stop_after_attempt(3),
    reraise=True,
)
def fail() -> None:
    raise ValueError("Invalid response")


try:
    fail()
except ValueError as exc:
    print(f"Operation failed: {exc}")

Logging Before a Retry

import logging

from tenacity import (
    before_sleep_log,
    retry,
    stop_after_attempt,
    wait_fixed,
)


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@retry(
    stop=stop_after_attempt(4),
    wait=wait_fixed(1),
    before_sleep=before_sleep_log(
        logger,
        logging.WARNING,
    ),
    reraise=True,
)
def unreliable_task() -> None:
    raise ConnectionError("Connection lost")

The callback runs after a failed attempt and before Tenacity sleeps for the next attempt.

Custom Retry Callback

Callbacks receive a RetryCallState object containing information about the current retry operation.

from tenacity import (
    RetryCallState,
    retry,
    stop_after_attempt,
    wait_fixed,
)


def report_retry(retry_state: RetryCallState) -> None:
    exception = retry_state.outcome.exception()

    print(
        f"Attempt {retry_state.attempt_number} failed: "
        f"{exception}"
    )


@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    before_sleep=report_retry,
    reraise=True,
)
def run_task() -> None:
    raise RuntimeError("Task failed")

Return a Fallback Value After Exhaustion

Use retry_error_callback to return a fallback value instead of raising after all attempts fail.

from typing import Optional

from tenacity import (
    RetryCallState,
    retry,
    stop_after_attempt,
    wait_fixed,
)


def return_none(
    retry_state: RetryCallState,
) -> Optional[str]:
    return None


@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    retry_error_callback=return_none,
)
def load_optional_value() -> str:
    raise ConnectionError("Source unavailable")


value = load_optional_value()
print(value)

Use fallbacks carefully so permanent failures are not hidden unintentionally.

Async Functions

The retry decorator also supports async def functions.

import asyncio

from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
)


@retry(
    stop=stop_after_attempt(5),
    wait=wait_random_exponential(
        multiplier=0.5,
        max=8,
    ),
    reraise=True,
)
async def fetch_async() -> str:
    await asyncio.sleep(0.1)
    raise ConnectionError("Async service unavailable")


async def main() -> None:
    try:
        await fetch_async()
    except ConnectionError as exc:
        print(f"Request failed: {exc}")


if __name__ == "__main__":
    asyncio.run(main())

Tenacity performs retry waits asynchronously for coroutine functions.

Retry a Code Block

Use Retrying when the retryable operation should remain inside an existing function.

from tenacity import Retrying, stop_after_attempt, wait_fixed


for attempt in Retrying(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    reraise=True,
):
    with attempt:
        print(
            f"Running attempt "
            f"{attempt.retry_state.attempt_number}"
        )
        raise ConnectionError("Temporary failure")

Async Retryable Code Block

from tenacity import (
    AsyncRetrying,
    stop_after_attempt,
    wait_fixed,
)


async def run_async_block() -> None:
    async for attempt in AsyncRetrying(
        stop=stop_after_attempt(3),
        wait=wait_fixed(1),
        reraise=True,
    ):
        with attempt:
            raise ConnectionError("Temporary failure")

Runtime Policy Overrides

A decorated function exposes retry_with, which can apply a different retry policy for one call.

from tenacity import retry, stop_after_attempt, wait_fixed


@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    reraise=True,
)
def process() -> None:
    raise RuntimeError("Processing failed")


process.retry_with(
    stop=stop_after_attempt(5),
    wait=wait_fixed(2),
)()

Retry Statistics

A decorated function exposes retry statistics through its retry attribute.

from tenacity import retry, stop_after_attempt


@retry(stop=stop_after_attempt(3))
def unstable() -> None:
    raise RuntimeError("Failure")


try:
    unstable()
except Exception:
    pass


print(unstable.retry.statistics)

Testing Retry-Decorated Functions

Override long waits in tests:

from tenacity import retry, stop_after_attempt, wait_fixed


@retry(
    stop=stop_after_attempt(5),
    wait=wait_fixed(10),
    reraise=True,
)
def external_operation() -> None:
    raise ConnectionError("Unavailable")


def test_external_operation() -> None:
    try:
        external_operation.retry_with(
            stop=stop_after_attempt(2),
            wait=wait_fixed(0),
        )()
    except ConnectionError:
        pass

You can also mock Tenacity's sleep behavior when a test requires more control.

Practical Example: Retrying an HTTP Request

Tenacity is independent of any particular HTTP client. The following example uses requests:

python -m pip install requests tenacity

import requests

from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_random_exponential,
)


TRANSIENT_STATUS_CODES = {
    429,
    500,
    502,
    503,
    504,
}


class RetryableHTTPError(Exception):
    pass


@retry(
    retry=retry_if_exception_type(
        (
            requests.Timeout,
            requests.ConnectionError,
            RetryableHTTPError,
        )
    ),
    stop=stop_after_attempt(5),
    wait=wait_random_exponential(
        multiplier=1,
        max=30,
    ),
    reraise=True,
)
def get_json(url: str) -> dict:
    response = requests.get(
        url,
        timeout=10,
    )

    if response.status_code in TRANSIENT_STATUS_CODES:
        raise RetryableHTTPError(
            f"Temporary HTTP status: "
            f"{response.status_code}"
        )

    response.raise_for_status()
    return response.json()

The request timeout and retry policy solve different problems:

• The timeout limits how long one request may block.
• The retry policy controls whether and when another request is attempted.

API Cheat Sheet

Component	Purpose
retry	Decorate a function with a retry policy
stop_after_attempt(n)	Stop after n total attempts
stop_after_delay(seconds)	Stop after an elapsed-time limit
wait_fixed(seconds)	Wait a constant amount of time
wait_random(min, max)	Wait for a random duration
wait_exponential(...)	Apply exponential backoff
wait_random_exponential(...)	Apply randomized exponential backoff
retry_if_exception_type(...)	Retry selected exception types
retry_if_result(predicate)	Retry selected return values
before_sleep_log(...)	Log failures before retrying
Retrying	Retry synchronous code or functions
AsyncRetrying	Retry asynchronous code
RetryCallState	Inspect the current retry invocation
RetryError	Default error after retry exhaustion
TryAgain	Request an explicit retry from inside a function

Recommended Practices

1. Retry only transient failures.
   Invalid input, authentication failures, and permission errors usually should not be retried.

2. Always define a stopping rule.
   Unbounded retries can cause stalled workers and hidden outages.

3. Use backoff and jitter for remote services.
   Immediate synchronized retries can make an outage worse.

4. Set operation-level timeouts.
   A retry policy cannot interrupt a network call that has no timeout.

5. Consider idempotency.
   Retrying a write operation may create duplicate records, charges, messages, or side effects.

6. Log retry exhaustion.
   Ensure permanent failures remain visible in monitoring and logs.

7. Respect server guidance.
   For HTTP 429 or similar responses, honor Retry-After when the service provides it.

When Not to Retry

Avoid automatic retries when:

• The operation is known to be non-idempotent
• The error is permanent or deterministic
• Retrying may duplicate a payment or external side effect
• The caller needs an immediate failure response
• The downstream service explicitly says not to retry
• A circuit breaker or queue is a better recovery mechanism

Resources

• Documentation: https://tenacity.readthedocs.io/
• Source code: https://github.com/jd/tenacity
• PyPI: https://pypi.org/project/tenacity/

License

Tenacity is distributed under the Apache License 2.0.

This independently written README may be adapted for your own project documentation.