The decorator is the easy part

A Python retry decorator is simple to write and surprisingly easy to make unsafe.

The usual first version catches every exception, sleeps for a fixed interval, and calls the function again:

import time
from functools import wraps


def retry(attempts: int = 3):
    def decorator(function):
        @wraps(function)
        def wrapper(*args, **kwargs):
            for attempt in range(attempts):
                try:
                    return function(*args, **kwargs)
                except Exception:
                    if attempt == attempts - 1:
                        raise
                    time.sleep(1)

        return wrapper

    return decorator

This preserves basic function metadata through functools.wraps, but almost everything else is underspecified.

It retries programming errors and invalid requests. It blocks the event loop when applied to async code. It has no operation deadline, jitter, result validation, hooks, or execution history. It also assumes that repeating the operation is safe.

A production retry decorator must make those decisions explicit rather than hide them behind @retry().

The manual examples in this guide were executed with Python 3.13.5 on September 13, 2026. Python’s current functools.wraps, ParamSpec, and asyncio cancellation semantics were checked against the official documentation on the same date.

What the decorator has to preserve

A useful retry abstraction needs two kinds of correctness.

The first is callable correctness:

  • preserve __name__, documentation, annotations, and the wrapped function;
  • preserve positional and keyword argument types;
  • return an accurate sync or async type;
  • never use time.sleep() inside an async wrapper;
  • propagate asyncio.CancelledError.

The second is operation correctness:

  • retry only classified exceptions or rejected results;
  • use capped backoff with jitter;
  • stop after bounded attempts and elapsed time;
  • expose attempts, delays, duration, and final outcome;
  • leave unsafe or ambiguous operations to a higher-level policy.

functools.wraps adds __wrapped__ and copies the metadata tools use to inspect the original callable. ParamSpec and TypeVar preserve its input and output types in the decorator signature.

A production-oriented core

The implementation below supports sync and async callables, exception and result classification, full-jitter exponential backoff, an elapsed-time budget, hooks, and a structured successful result.

Failure is not converted into a false success. Exhaustion raises RetryExhausted with the complete attempt history.

retry_decorator.py
from __future__ import annotations

import asyncio
import inspect
import random
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, replace
from functools import wraps
from typing import Generic, Literal, ParamSpec, TypeVar, overload


P = ParamSpec("P")
R = TypeVar("R")
Outcome = Literal["failed", "rejected_result", "succeeded"]


@dataclass(frozen=True)
class Attempt:
    number: int
    outcome: Outcome
    duration_seconds: float
    delay_seconds: float = 0.0
    error_type: str | None = None


@dataclass(frozen=True)
class RetryResult(Generic[R]):
    value: R
    attempts: tuple[Attempt, ...]
    elapsed_seconds: float


class RetryExhausted(RuntimeError):
    def __init__(
        self,
        message: str,
        *,
        attempts: tuple[Attempt, ...],
    ) -> None:
        super().__init__(message)
        self.attempts = attempts


@dataclass(frozen=True)
class RetryConfig(Generic[R]):
    max_attempts: int = 3
    retry_on: tuple[type[Exception], ...] = (ConnectionError,)
    retry_if_result: Callable[[R], bool] | None = None
    base_delay: float = 0.25
    max_delay: float = 3.0
    max_elapsed: float = 10.0
    before_attempt: Callable[[int], None] | None = None
    on_error: Callable[[Attempt], None] | None = None
    on_complete: Callable[[RetryResult[R]], None] | None = None

    def delay(
        self,
        retry_number: int,
        *,
        uniform: Callable[[float, float], float],
    ) -> float:
        ceiling = min(
            self.max_delay,
            self.base_delay * 2 ** (retry_number - 1),
        )
        return uniform(0.0, ceiling)


@overload
def retry(
    config: RetryConfig[R],
    *,
    uniform: Callable[[float, float], float] = random.uniform,
    sleep: Callable[[float], None] = time.sleep,
    clock: Callable[[], float] = time.monotonic,
) -> Callable[
    [Callable[P, R]],
    Callable[P, RetryResult[R]],
]: ...


@overload
def retry(
    config: RetryConfig[R],
    *,
    uniform: Callable[[float, float], float] = random.uniform,
    async_sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
    clock: Callable[[], float] = time.monotonic,
) -> Callable[
    [Callable[P, Awaitable[R]]],
    Callable[P, Awaitable[RetryResult[R]]],
]: ...


def retry(
    config: RetryConfig[R],
    *,
    uniform: Callable[[float, float], float] = random.uniform,
    sleep: Callable[[float], None] = time.sleep,
    async_sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
    clock: Callable[[], float] = time.monotonic,
):
    def decorator(function):
        if inspect.iscoroutinefunction(function):
            @wraps(function)
            async def async_wrapper(*args, **kwargs):
                attempts: list[Attempt] = []
                started = clock()

                async with asyncio.timeout(config.max_elapsed):
                    for number in range(1, config.max_attempts + 1):
                        if config.before_attempt:
                            config.before_attempt(number)

                        attempt_started = clock()

                        try:
                            value = await function(*args, **kwargs)
                        except asyncio.CancelledError:
                            raise
                        except config.retry_on as exc:
                            record = Attempt(
                                number=number,
                                outcome="failed",
                                duration_seconds=clock() - attempt_started,
                                error_type=type(exc).__name__,
                            )
                            attempts.append(record)

                            if config.on_error:
                                config.on_error(record)

                            if number == config.max_attempts:
                                raise RetryExhausted(
                                    "Retry attempts exhausted",
                                    attempts=tuple(attempts),
                                ) from exc
                        else:
                            if not (
                                config.retry_if_result
                                and config.retry_if_result(value)
                            ):
                                record = Attempt(
                                    number=number,
                                    outcome="succeeded",
                                    duration_seconds=clock() - attempt_started,
                                )
                                result = RetryResult(
                                    value=value,
                                    attempts=tuple((*attempts, record)),
                                    elapsed_seconds=clock() - started,
                                )

                                if config.on_complete:
                                    config.on_complete(result)

                                return result

                            attempts.append(
                                Attempt(
                                    number=number,
                                    outcome="rejected_result",
                                    duration_seconds=clock() - attempt_started,
                                )
                            )

                        delay = config.delay(number, uniform=uniform)
                        attempts[-1] = replace(
                            attempts[-1],
                            delay_seconds=delay,
                        )
                        await async_sleep(delay)

                raise AssertionError("unreachable")

            return async_wrapper

        @wraps(function)
        def sync_wrapper(*args, **kwargs):
            attempts: list[Attempt] = []
            started = clock()

            for number in range(1, config.max_attempts + 1):
                if clock() - started >= config.max_elapsed:
                    raise RetryExhausted(
                        "Retry deadline exhausted",
                        attempts=tuple(attempts),
                    )

                if config.before_attempt:
                    config.before_attempt(number)

                attempt_started = clock()

                try:
                    value = function(*args, **kwargs)
                except config.retry_on as exc:
                    record = Attempt(
                        number=number,
                        outcome="failed",
                        duration_seconds=clock() - attempt_started,
                        error_type=type(exc).__name__,
                    )
                    attempts.append(record)

                    if config.on_error:
                        config.on_error(record)

                    if number == config.max_attempts:
                        raise RetryExhausted(
                            "Retry attempts exhausted",
                            attempts=tuple(attempts),
                        ) from exc
                else:
                    if not (
                        config.retry_if_result
                        and config.retry_if_result(value)
                    ):
                        record = Attempt(
                            number=number,
                            outcome="succeeded",
                            duration_seconds=clock() - attempt_started,
                        )
                        result = RetryResult(
                            value=value,
                            attempts=tuple((*attempts, record)),
                            elapsed_seconds=clock() - started,
                        )

                        if config.on_complete:
                            config.on_complete(result)

                        return result

                    attempts.append(
                        Attempt(
                            number=number,
                            outcome="rejected_result",
                            duration_seconds=clock() - attempt_started,
                        )
                    )

                delay = config.delay(number, uniform=uniform)
                attempts[-1] = replace(
                    attempts[-1],
                    delay_seconds=delay,
                )

                if clock() - started + delay >= config.max_elapsed:
                    raise RetryExhausted(
                        "Retry deadline exhausted",
                        attempts=tuple(attempts),
                    )

                sleep(delay)

            raise AssertionError("unreachable")

        return sync_wrapper

    return decorator

The sync deadline is cooperative. It stops another attempt from starting, but cannot interrupt a blocking function already in progress. Hard sync cancellation needs a process, thread, or transport boundary with its own timeout.

The async deadline uses asyncio.timeout() around the complete operation, including attempts and sleeps. Cancellation is propagated rather than treated as provider failure.

Test decisions without waiting

Randomness and sleeping are injected, so tests can use deterministic values.

test_retry_decorator.py
def test_sync_retry_exposes_attempt_history() -> None:
    calls = 0
    delays: list[float] = []

    @retry(
        RetryConfig[str](
            max_attempts=3,
            retry_on=(ConnectionError,),
        ),
        uniform=lambda low, high: high / 2,
        sleep=delays.append,
    )
    def load_value() -> str:
        nonlocal calls
        calls += 1

        if calls < 3:
            raise ConnectionError("temporary")

        return "ok"

    result = load_value()

    assert result.value == "ok"
    assert calls == 3
    assert delays == [0.125, 0.25]
    assert [attempt.outcome for attempt in result.attempts] == [
        "failed",
        "failed",
        "succeeded",
    ]
    assert load_value.__name__ == "load_value"

Async tests should also cancel during backoff and confirm that no later attempt starts. Test rejected results separately from exceptions because they represent different failure semantics.

For a deeper comparison of jitter and total budgets, read Making Python Exponential Backoff Safe with Jitter, Retry Budgets and Production Trade-offs.

The RelPrim decorator keeps the history explicit

RelPrim’s @resilient decorator is async-first and uses the same operation model as its explicit builder API.

relprim_decorator.py
from relprim import (
    ExponentialBackoff,
    RetryPolicy,
    TimeoutPolicy,
    resilient,
    validation_policy,
    validator,
)


class TemporaryProviderError(RuntimeError):
    pass


non_empty = validation_policy(
    validator(
        "non_empty_output",
        lambda value: bool(value.strip()),
        message="Provider output must not be empty.",
    )
)


@resilient(
    name="generate_product_summary",
    retry=RetryPolicy(
        max_attempts=3,
        retry_on=(TemporaryProviderError,),
        backoff=ExponentialBackoff(
            base_delay_seconds=0.25,
            max_delay_seconds=3.0,
            jitter=True,
        ),
    ),
    timeout=TimeoutPolicy(seconds=10),
    validation=non_empty,
)
async def generate_summary(product: Product) -> str:
    return await provider.generate(product)


result = await generate_summary(product)

print(result.value)
print(result.report.to_dict())

The decorated function returns OperationResult[T], not only T. Attempts, durations, validation, fallback use, idempotency status, and the final operation outcome remain available in the report.

The simple form @resilient(retries=3) means three retries after the first attempt. The explicit RetryPolicy(max_attempts=3) counts the first execution inside those three attempts. Production code should prefer the explicit form when that distinction matters.

RelPrim preserves the decorated function with functools.wraps and types its arguments through ParamSpec. Its decorator currently targets async callables; the lower-level RetryPolicy supports both sync and async execution.

See the RelPrim getting started guide for the decorator API.

When the builder is clearer

A decorator works well when the policy is static and belongs naturally to one function.

RelPrim also exposes an explicit operation builder for cases where the policy is assembled dynamically, events need a dedicated sink, several fallbacks have names, or the same callable participates in different execution policies.

result = await (
    async_operation("generate_product_summary", generate_summary_raw)
    .with_retry(retry_policy)
    .with_timeout(timeout_policy)
    .with_validation(non_empty)
    .with_fallbacks(fallbacks)
    .with_events(events)
    .run(product)
)

This uses the same RelPrim execution model as @resilient, but keeps every policy decision visible at the call site.

The decorator optimizes adoption. The RelPrim builder optimizes visibility.

That trade-off is especially important around HTTP integrations, where method semantics, status codes, idempotency, and provider guidance may change the policy. See Building Safe HTTPX Retry Policies in Python with Backoff, Timeouts and Async Failure Handling and Building a Safe Python Requests Retry Policy with Backoff, Retry-After and Failure Handling.

Before you ship

Before adding a retry decorator, define the transient failures and rejected results it may repeat. Confirm that the operation is safe to replay and preserve one operation identity when a timeout can leave the remote outcome unknown.

Use functools.wraps, preserve types with ParamSpec, separate sync and async sleeping, propagate cancellation, and bound both attempts and elapsed time. Inject the clock, random source, and sleep functions so tests verify decisions without real delays.

Expose the attempt history and final outcome. A decorator should reduce repeated boilerplate, not hide the execution policy from callers and operators.

A retry decorator is useful when it makes behavior easier to see. Once it makes the operation harder to explain, use an explicit builder instead.

Where has a retry decorator become too magical in your codebase: exception classification, async cancellation, result validation, or observability?