tennacity 1.2.2

A module that prints Hello World on import

tennacity

Reliable and configurable retry behavior for Python.

tennacity 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 tennacity releases and Python 3.10 or newer.

Installation

Install tennacity from PyPI:

python -m pip install tennacity

Verify the installation:

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

Quick Start

from tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity sleeps for the next attempt.

Custom Retry Callback

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

from tennacity 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 tennacity 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 tennacity 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())

tennacity performs retry waits asynchronously for coroutine functions.

Retry a Code Block

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

from tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity 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 tennacity's sleep behavior when a test requires more control.

Practical Example: Retrying an HTTP Request

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

python -m pip install requests tennacity

import requests

from tennacity 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://tennacity.readthedocs.io/
• Source code: https://github.com/jd/tennacity
• PyPI: https://pypi.org/project/tennacity/

License

tennacity is distributed under the Apache License 2.0.

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