Python decorators: write a retry decorator for flaky calls.

A decorator is a callable that takes a function and returns a replacement, usually a wrapper that adds behaviour before or after calling the original. @retry(...) above a function is shorthand for func = retry(...)(func). Because the retry decorator needs configuration (attempts, delay, which exceptions), it is a decorator factory: a function that returns the decorator.

Implementation

import functools
import logging
import random
import time
from collections.abc import Callable
from typing import ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")
log = logging.getLogger(__name__)


def retry(
    *,
    attempts: int = 3,
    delay: float = 0.5,
    backoff: float = 2.0,
    max_delay: float = 10.0,
    exceptions: tuple[type[BaseException], ...] = (Exception,),
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    if attempts < 1:
        raise ValueError("attempts must be >= 1")

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            wait = delay
            for attempt in range(1, attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as exc:
                    if attempt == attempts:
                        raise  # re-raise the last error with its traceback
                    sleep_for = min(wait, max_delay) * random.uniform(0.5, 1.0)
                    log.warning("%s failed (%s), attempt %d/%d, retrying in %.2fs",
                                func.__name__, exc, attempt, attempts, sleep_for)
                    time.sleep(sleep_for)
                    wait *= backoff
            raise AssertionError("unreachable")
        return wrapper
    return decorator

Points worth calling out:

• Three levels: retry(...) receives options, decorator(func) receives the function, wrapper(*args, **kwargs) runs on every call.
• functools.wraps copies __name__, __doc__ and __wrapped__, so logs, pytest output and introspection show the real function.
• Only listed exceptions are retried. Defaulting to Exception is convenient but too broad for real use; pass the transient ones explicitly. An AssertionError raised by a check should normally not be retried.
• Bare raise re-raises the original exception with its traceback intact.
• Exponential backoff with a cap and jitter avoids hammering a struggling service and synchronised retries from parallel workers.
• ParamSpec (Python 3.10+) keeps the wrapped function’s signature visible to type checkers.

Using it

import requests

class TransientHTTPError(Exception):
    pass

@retry(attempts=4, delay=0.2, exceptions=(requests.ConnectionError, requests.Timeout, TransientHTTPError))
def get_order(session: requests.Session, order_id: str) -> dict:
    res = session.get(f"https://api.example.com/orders/{order_id}", timeout=5)
    if res.status_code in (429, 502, 503, 504):
        raise TransientHTTPError(f"HTTP {res.status_code}")
    res.raise_for_status()   # 4xx errors are not retried
    return res.json()

Async version

For async def functions, the wrapper must be async and use await asyncio.sleep(...); calling time.sleep would block the event loop. A production decorator often checks inspect.iscoroutinefunction(func) and returns the right wrapper.

Don’t reinvent in production

Libraries such as tenacity (general retries) and pytest-rerunfailures (re-running failed tests) are mature. Writing one in an interview shows you understand them; in a framework, prefer the library unless you need something tiny. And remember: retrying a flaky test hides the flakiness unless the retries are reported.

What interviewers look for

• Correct three-level structure and functools.wraps.
• Narrow exception handling and re-raising on the final attempt.
• Backoff, cap and jitter, with a reason for each.
• Judgement about what should not be retried and about existing libraries.