pythonintermediate

Python Functools and Decorator Patterns

Useful decorator patterns with functools including caching, retry, timing, and rate limiting.

python
import functools
import time
from typing import Callable, TypeVar, ParamSpec

P = ParamSpec("P")
R = TypeVar("R")


# Retry decorator with exponential backoff
def retry(max_attempts: int = 3, delay: float = 1.0):
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts:
                        raise
                    wait = delay * (2 ** (attempt - 1))
                    print(f"Attempt {attempt} failed: {e}. Retrying in {wait}s")
                    time.sleep(wait)
            raise RuntimeError("Unreachable")
        return wrapper
    return decorator


# Timing decorator
def timed(func: Callable[P, R]) -> Callable[P, R]:
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper


# Memoize with TTL
def ttl_cache(seconds: int = 300):
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        cache: dict[str, tuple[float, R]] = {}

        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            key = str((args, sorted(kwargs.items())))
            now = time.time()
            if key in cache and now - cache[key][0] < seconds:
                return cache[key][1]
            result = func(*args, **kwargs)
            cache[key] = (now, result)
            return result
        return wrapper
    return decorator


# Usage
@retry(max_attempts=3, delay=0.5)
@timed
def fetch_data(url: str) -> str:
    import urllib.request
    with urllib.request.urlopen(url) as resp:
        return resp.read().decode()


@ttl_cache(seconds=60)
def get_config() -> dict:
    return {"key": "value"}  # expensive operation

Use Cases

  • Adding retry logic to flaky operations
  • Performance profiling with timing decorators
  • Caching expensive computations with expiry

Tags

Related Snippets

Similar patterns you can reuse in the same workflow.