Decorator (add behavior to a function or object without changing its code)

When to use

• You want to add cross-cutting behavior—metrics, logging, tracing, caching, auth checks—around existing calls.
• You need to toggle features per function or per call site without editing the function body.
• You prefer composition over inheritance for small, orthogonal concerns.

Avoid when a single helper call inside the function is enough, or when behavior belongs in a separate layer (e.g., Proxy around an object).

Diagram (text)

Caller ──> [decorated] load_to_warehouse(...)
              ▲
     metrics decorator wraps the function
     (start timer → call → emit status/latency)

Python example (≤40 lines, type-hinted)

Concrete case: time a pipeline step and emit success/error with tags.

from __future__ import annotations
from typing import Callable, TypeVar, ParamSpec, Mapping
from functools import wraps
import time

P = ParamSpec("P"); T = TypeVar("T")
Sink = Callable[[str, float, Mapping[str, str]], None]

def metrics(name: str, sink: Sink, **base_tags: str) -> Callable[[Callable[P, T]], Callable[P, T]]:
    def deco(fn: Callable[P, T]) -> Callable[P, T]:
        @wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
            t0 = time.time()
            try:
                result = fn(*args, **kwargs)
                sink(name, time.time() - t0, {**base_tags, "status": "ok"})
                return result
            except Exception:
                sink(name, time.time() - t0, {**base_tags, "status": "error"})
                raise
        return wrapper
    return deco

# Example: decorate a loader step
def load_to_warehouse(rows: int) -> int:
    if rows <= 0: raise ValueError("no data")
    return rows

events: list[tuple[str, float, dict[str, str]]] = []
@metrics("load", lambda n, d, t: events.append((n, d, dict(t))), target="snowflake")
def safe_load(rows: int) -> int:
    return load_to_warehouse(rows)

Tiny pytest (cements it)

def test_metrics_decorator_records_ok_and_error():
    events.clear()
    assert safe_load(5) == 5
    assert events[-1][0] == "load" and events[-1][2]["status"] == "ok"
    try: safe_load(0)
    except ValueError: pass
    assert events[-1][2]["status"] == "error" and events[-1][2]["target"] == "snowflake"

Trade-offs & pitfalls

• Pros: Zero changes to core logic; reusable; stackable (e.g., @metrics then @retry); great for A/B enabling.
• Cons: Layers can hide flow; debugging stack traces can be noisier.
• Pitfalls:
  • Forgetting @wraps(fn) → breaks name, docstring, and signature for tools.
  • Not handling async functions—sync wrapper won’t await; write an async version when needed.
  • Decorating methods that expect self—works fine, but mind order when stacking decorators.
  • Swallowing exceptions—emit metrics but re-raise.

Pythonic alternatives

• GoF Decorator (object wrapper): wrap objects to add features at the same interface (very close to Proxy).
• Context managers for scoped timing/logging: with timed("load"): ....
• functools.lru_cache / cachetools for pure caching; clearer than a custom decorator.
• Middleware at service boundaries (FastAPI/ASGI) for request-level concerns instead of per-function.

Mini exercise

Write an async version: async_metrics(...) that detects/awaits coroutine functions and emits the same events. Add a test decorating an async def step() that raises once and succeeds once.

Checks (quick checklist)

• Use @wraps to preserve metadata/signature.
• Keep the decorator small and orthogonal (no business logic).
• Don’t swallow errors; re-raise after side effects.
• Consider async and method cases if applicable.
• Tests verify both success and failure paths, and that tags are emitted.