Decorators Done Right

Introduction

Welcome to the third course in the Advanced Python Language Features path! By completing the previous two courses on the Python Data Model and Class Machinery, you've built a solid foundation in designing custom types, implementing protocols, and creating extensible systems. Now, we're ready to explore functional programming patterns and modern control flow structures that will make your code more expressive and declarative. In this course, we'll master four essential patterns: decorators, generic functions with single dispatch, structural pattern matching, and composable error handling. Today's lesson focuses on decorators done right : we'll build a configurable retry mechanism with exponential backoff and jitter, learning how to preserve function metadata, control exception handling, and leverage advanced typing features for type-safe decorator factories. By the end of this lesson, you'll be able to write decorators that feel professional and production-ready.

Why Decorators Matter

Decorators are powerful tools for adding reusable behavior to functions without modifying their core logic. However, many decorators in the wild have issues: they lose function metadata, break type checking, or lack configurability. A production-grade decorator must handle these concerns (and potentially more) gracefully. Consider a retry decorator for network operations. A naive implementation might catch all exceptions and retry indefinitely, which could mask permanent failures or overwhelm a failing service. A robust solution needs: Exponential backoff to reduce load on struggling systems Jitter to prevent synchronized retry storms Selective exception handling to distinguish transient from permanent failures Preserved function identity so debugging and introspection work correctly Type safety so IDEs and type checkers understand the decorated function These requirements transform a simple wrapper into a sophisticated, reusable component.

The Basic Retry Pattern

Let's start with the simplest possible retry decorator to establish the core pattern. A retry decorator wraps a function and catches exceptions, attempting the call again after a delay: Python import time from functools import wraps def simple_retry(func): @wraps(func) def wrapper(*args, **kwargs): for attempt in range(3): try: return func(*args, **kwargs) except Exception: if attempt == 2: raise time.sleep(1.0) return None # unreachable, but satisfies type checker return wrapper This basic decorator tries the function up to three times, waiting one second between attempts. The @wraps(func) line preserves the original function's metadata. However, this implementation is inflexible: the retry count and delay are hardcoded, and it catches all exceptions indiscriminately.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Implementing the Retry Logic

Now let's implement the core retry loop with exponential backoff and jitter. The wrapper needs to track attempts and calculate increasing delays:

Python

import random
import time
from typing import cast

def retry(
    *,
    retries: int = 3,
    delay: float = 0.05,
    backoff: float = 2.0,
    jitter: float = 0.2,
    exceptions: tuple[type[BaseException], ...] = (Exception,),
    predicate: Callable[[BaseException], bool] | None = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    def decorator(func: Callable[P, T]) -> Callable[P, T]:
        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
            attempt = 0
            while True:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if predicate is not None and not predicate(e):
                        raise
                    if attempt >= retries:
                        raise
                    base = delay * (backoff ** attempt)
                    sleep_for = base + (random.uniform(0, base * jitter) if jitter else 0.0)
                    time.sleep(max(0.0, sleep_for))
                    attempt += 1
        return cast(Callable[P, T], wrapper)
    return decorator

import random
import time
from typing import cast

def retry(
    *,
    retries: int = 3,
    delay: float = 0.05,
    backoff: float = 2.0,
    jitter: float = 0.2,
    exceptions: tuple[type[BaseException], ...] = (Exception,),
    predicate: Callable[[BaseException], bool] | None = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    def decorator(func: Callable[P, T]) -> Callable[P, T]:
        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
            attempt = 0
            while True:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if predicate is not None and not predicate(e):
                        raise
                    if attempt >= retries:
                        raise
                    base = delay * (backoff ** attempt)
                    sleep_for = base + (random.uniform(0, base * jitter) if jitter else 0.0)
                    time.sleep(max(0.0, sleep_for))
                    attempt += 1
        return cast(Callable[P, T], wrapper)
    return decorator

The retry logic works as follows:

We attempt the function call inside a try block
If it succeeds, we return immediately
If it raises one of the specified exceptions, we check the predicate (if provided)
If the predicate returns False or we've exhausted retries, we re-raise the exception
Otherwise, we calculate a sleep duration using exponential backoff and jitter, then try again

The cast(Callable[P, T], wrapper) helps the type checker understand that wrapper matches the signature of the decorated function. Note that this decorator implementation is designed for synchronous functions: it won't work correctly with async functions because it uses time.sleep() (which blocks the event loop) and doesn't handle awaitable return values. But don't worry, we'll be encountering asynchronous functions soon enough, in the next course of this path!

Testing the Decorator in Practice

Let's see the decorator in action with a function that succeeds after a specific number of attempts:

Python

def make_unstable_svc(success_after: int) -> Callable[[], str]:
    calls = {"n": 0}

    @retry(retries=5, delay=0.01, backoff=2.0, jitter=0.1, exceptions=(TransientError,))
    def unstable() -> str:
        calls["n"] += 1
        if calls["n"] < success_after:
            raise TransientError("temporary failure", retryable=True)
        return f"OK-{calls['n']}"
    unstable.calls = calls  # type: ignore[attr-defined]
    return unstable

svc = make_unstable_svc(success_after=3)
print(svc())
print(svc.__name__)
print(svc.calls["n"])  # type: ignore[index]

def make_unstable_svc(success_after: int) -> Callable[[], str]:
    calls = {"n": 0}

    @retry(retries=5, delay=0.01, backoff=2.0, jitter=0.1, exceptions=(TransientError,))
    def unstable() -> str:
        calls["n"] += 1
        if calls["n"] < success_after:
            raise TransientError("temporary failure", retryable=True)
        return f"OK-{calls['n']}"
    unstable.calls = calls  # type: ignore[attr-defined]
    return unstable

svc = make_unstable_svc(success_after=3)
print(svc())
print(svc.__name__)
print(svc.calls["n"])  # type: ignore[index]

text

OK-3
unstable
3

OK-3
unstable
3

The service fails twice with TransientError, then succeeds on the third call. Notice that svc.__name__ is "unstable", not "wrapper", demonstrating that @wraps preserved the function's identity. The call count confirms three attempts were needed.

Now let's test permanent failures and predicate-based filtering:

Python

try:
    permanent_failure()
except Exception as e:
    print(type(e).__name__)

flips = [True, False, True]
try:
    sometimes_nonretryable(flips)
except Exception as e:
    print(str(e))

try:
    permanent_failure()
except Exception as e:
    print(type(e).__name__)

flips = [True, False, True]
try:
    sometimes_nonretryable(flips)
except Exception as e:
    print(str(e))

text

PermanentError
fatal

PermanentError
fatal

The permanent_failure function raises PermanentError, which isn't in the exceptions tuple, so it propagates immediately without retries. The sometimes_nonretryable function raises a TransientError with retryable=False on the second call, and the predicate causes it to propagate immediately, showing the message "fatal".

Finally, let's verify that the decorator respects the retry limit:

Python

fail_fast = make_unstable_svc(success_after=999)
try:
    fail_fast()
except Exception as e:
    print(type(e).__name__)

fail_fast = make_unstable_svc(success_after=999)
try:
    fail_fast()
except Exception as e:
    print(type(e).__name__)

text

TransientError

TransientError

Even though the service would eventually succeed, it never will within our 5-retry limit, so the exception is re-raised after the final attempt.

Introduction

Why Decorators Matter

The Basic Retry Pattern

Join the 1M+ learners on CodeSignal

Preserving Function Identity

Type-Safe Decorators with ParamSpec

Building a Configurable Retry Factory

Implementing the Retry Logic

Exponential Backoff and Jitter Explained

Controlling Which Exceptions to Retry

Conditional Retries with Predicates

Testing the Decorator in Practice

Conclusion and Next Steps