Decoradores en producción: patrones reales en Python

Los decoradores brillan cuando tienes una preocupación transversal —algo que no pertenece a la lógica de negocio pero que necesita envolver decenas de funciones: caché, telemetría, reintentos, validación. La stdlib y los frameworks populares ya vienen cargados de ellos, y entender cómo funcionan por dentro te permite escribir los tuyos con criterio.

Antes del primer bloque de código, vale la pena entender por qué Python eligió esta sintaxis. Un decorador no es magia: es azúcar sintáctica sobre la llamada func = decorator(func). El objeto función pasa como argumento, el decorador devuelve un callable nuevo (normalmente un wrapper) que cierra sobre el original, y ese wrapper reemplaza al nombre en el módulo. Eso es todo. La consecuencia importante es que si olvidas @functools.wraps(func) en el wrapper, pierdes __name__, __doc__ y __wrapped__, lo que rompe introspección, logging y herramientas como help().

¿Cuándo usarlos? Cuando el comportamiento extra es ortogonal a lo que hace la función: medir su tiempo no cambia su contrato, reintentar una llamada HTTP no cambia su interfaz, cachear no cambia su semántica (asumiendo que es pura). Si el comportamiento está entrelazado con la lógica —si el decorador necesita saber qué hace la función para decidir cómo actuar—, probablemente es mejor composición explícita.

¿Qué rompe si lo haces mal? Las tres trampas clásicas: olvidar wraps, mutar estado compartido en el closure sin protección de hilos, y aplicar caché a funciones con efectos secundarios.

import functools
import logging
import time
import random
from typing import Any, Callable, TypeVar

F = TypeVar("F", bound=Callable[..., Any])

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(name)s — %(message)s",
)
logger = logging.getLogger(__name__)


# ── 1. Timing ────────────────────────────────────────────────────────────────

def timed(func: F) -> F:
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()          # resolución sub-microsegundo
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        logger.debug("%s completó en %.4fs", func.__qualname__, elapsed)
        return result
    return wrapper  # type: ignore[return-value]


# ── 2. Logging de entradas, salidas y excepciones ────────────────────────────

def logged(func: F) -> F:
    fn_logger = logging.getLogger(func.__module__)

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        fn_logger.debug("→ %s args=%r kwargs=%r", func.__qualname__, args, kwargs)
        try:
            result = func(*args, **kwargs)
        except Exception as exc:
            fn_logger.exception("✗ %s lanzó %r", func.__qualname__, exc)
            raise                            # re-raise para no silenciar el error
        fn_logger.debug("← %s retornó %r", func.__qualname__, result)
        return result
    return wrapper  # type: ignore[return-value]


# ── 3. Retry con backoff exponencial ─────────────────────────────────────────

def retry(
    attempts: int = 3,
    base_delay: float = 0.5,
    exceptions: tuple[type[Exception], ...] = (Exception,),
):
    """Decorador parametrizado: retry(attempts=5)(func)."""
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as exc:
                    if attempt == attempts:
                        raise              # agotamos los intentos, propagamos
                    delay = base_delay * (2 ** (attempt - 1))  # backoff exponencial
                    logger.warning(
                        "%s falló (intento %d/%d): %r — reintentando en %.2fs",
                        func.__qualname__, attempt, attempts, exc, delay,
                    )
                    time.sleep(delay)
            return None  # nunca se alcanza; satisface type checkers
        return wrapper  # type: ignore[return-value]
    return decorator


# ── 4. Validación de argumentos ───────────────────────────────────────────────

def require_positive(*param_names: str):
    """Verifica que los argumentos nombrados sean > 0 antes de llamar la función."""
    def decorator(func: F) -> F:
        sig = __import__("inspect").signature(func)   # construido una sola vez

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            bound = sig.bind(*args, **kwargs)
            bound.apply_defaults()
            for name in param_names:
                if name in bound.arguments and bound.arguments[name] <= 0:
                    raise ValueError(f"'{name}' debe ser positivo, recibido {bound.arguments[name]!r}")
            return func(*args, **kwargs)
        return wrapper  # type: ignore[return-value]
    return decorator


# ── 5. Caché con lru_cache / cache ────────────────────────────────────────────

@functools.lru_cache(maxsize=128)   # LRU clásico, descarta los menos usados
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)


@functools.cache                    # desde 3.9: unbounded, más rápido que lru_cache(None)
def expensive_lookup(key: str) -> str:
    time.sleep(0.01)                # simula I/O
    return key.upper()


# ── 6. @property como decorador del modelo de datos ──────────────────────────

class Circle:
    def __init__(self, radius: float) -> None:
        self._radius = radius

    @property
    def radius(self) -> float:
        return self._radius

    @radius.setter
    def radius(self, value: float) -> None:
        if value < 0:
            raise ValueError("El radio no puede ser negativo")
        self._radius = value

    @property
    def area(self) -> float:
        import math
        return math.pi * self._radius ** 2


# ── 7. Mini-framework de rutas (cómo funciona Flask/FastAPI internamente) ─────

class Router:
    def __init__(self) -> None:
        self._routes: dict[tuple[str, str], Callable] = {}

    def route(self, path: str, method: str = "GET"):
        """Registra la función en el diccionario de rutas al decorar."""
        def decorator(func: Callable) -> Callable:
            self._routes[(method.upper(), path)] = func
            return func              # devuelve la función sin modificarla
        return decorator

    def dispatch(self, method: str, path: str, **kwargs):
        handler = self._routes.get((method.upper(), path))
        if handler is None:
            raise KeyError(f"Ruta no encontrada: {method} {path}")
        return handler(**kwargs)


app = Router()

@app.route("/users", method="GET")
@timed
@logged
def list_users(limit: int = 10):
    return [f"user_{i}" for i in range(limit)]


@app.route("/transfer", method="POST")
@retry(attempts=4, base_delay=0.1, exceptions=(ConnectionError, TimeoutError))
@require_positive("amount")
@logged
def transfer_funds(account_from: str, account_to: str, amount: float):
    if random.random() < 0.5:      # simula fallo intermitente
        raise ConnectionError("servicio de pagos no disponible")
    return {"status": "ok", "amount": amount}


# ── Ejecución ─────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    # Cache
    print(fibonacci(30))
    print(fibonacci.cache_info())   # CacheInfo(hits=..., misses=31, ...)

    # Ruta GET
    result = app.dispatch("GET", "/users", limit=3)
    print(result)

    # Ruta POST con retry y validación
    try:
        result = app.dispatch(
            "POST", "/transfer",
            account_from="A1", account_to="B2", amount=250.0,
        )
        print(result)
    except ConnectionError:
        print("Todos los intentos fallaron")

Qué está pasando y por qué importa cada decisión

@timed usa time.perf_counter() en lugar de time.time() porque perf_counter tiene la mayor resolución disponible en el sistema y no está sujeto a ajustes del reloj del sistema (NTP, DST). Para medir latencia de funciones, es la elección correcta siempre.

@logged captura la excepción, la registra con logger.exception (que incluye el traceback completo automáticamente) y hace raise sin argumentos. Ese raise desnudo re-lanza la excepción original conservando el traceback completo, a diferencia de raise exc que lo truncaría.

@retry es un decorador parametrizado: cuando escribes @retry(attempts=4), Python ejecuta retry(attempts=4) que devuelve decorator, y ese decorator es el que se aplica a la función. Por eso hay tres niveles de anidamiento. El backoff exponencial —base_delay * 2^(attempt-1)— evita saturar el servicio que está fallando; con base_delay=0.5 los tiempos de espera serían 0.5s, 1s, 2s, 4s.

@require_positive construye el objeto inspect.Signature una sola vez en tiempo de decoración, no en cada llamada. sig.bind(*args, **kwargs) + apply_defaults() normaliza tanto argumentos posicionales como keyword al mismo diccionario, así el decorador funciona independientemente de cómo el caller pase los argumentos.

@functools.cache (Python 3.9+) es equivalente a lru_cache(maxsize=None) pero más rápido internamente porque no tiene que gestionar la estructura LRU. Úsalo cuando el conjunto de claves es acotado y predecible. Si no lo es, lru_cache(maxsize=N) te da control de memoria.

@property es simplemente el descriptor property usado como decorador. Cuando el intérprete evalúa @property, llama property(radius) que devuelve un objeto descriptor que intercepta __get__ y __set__. @radius.setter llama radius.setter(func), devolviendo un nuevo descriptor con el setter añadido. No hay nada especial en la sintaxis: es el mismo protocolo de decoradores.

El mini-framework de rutas muestra exactamente lo que hace @app.route("/users") en Flask: registra la función en un diccionario interno {(method, path): handler} y devuelve la función sin modificarla. Flask añade más bookkeeping (blueprints, URL converters, etc.), pero el núcleo es ese diccionario. El handler en sí permanece invocable directamente, lo que facilita testearlo sin levantar un servidor.

La pila de decoradores en list_users@timed sobre @logged— se aplica de abajo a arriba: primero logged envuelve list_users, luego timed envuelve el resultado. La ejecución fluye de afuera hacia adentro, así que timed mide el tiempo total incluyendo el overhead de logging, lo cual suele ser lo que quieres en producción.

Errores que debes conocer

Error: olvidar @functools.wraps en el wrapper hace que todas las funciones decoradas reportensu nombre como wrapper, rompiendo logs, profilers y el atributo __name__.

# ❌ Wrong
def timed(func):
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        return func(*args, **kwargs)
    return wrapper

print(list_users.__name__)  # 'wrapper'

# ✅ Right
def timed(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        return func(*args, **kwargs)
    return wrapper

print(list_users.__name__)  # 'list_users'

functools.wraps copia __name__, __qualname__, __doc__, __annotations__ y __wrapped__; este último permite a herramientas como inspect.unwrap() traversar la cadena de decoradores.

Error: aplicar @lru_cache a métodos de instancia hace que self sea parte de la clave, creando una referencia fuerte que impide que las instancias sean recolectadas por el GC.

# ❌ Wrong
class Processor:
    @functools.lru_cache(maxsize=256)
    def compute(self, value: int) -> int:   # self queda retenido en el caché
        return value * 2

# ✅ Right
class Processor:
    @functools.cached_property              # caché por instancia, se libera con ella
    def config(self):                       # para propiedades costosas sin argumentos
        return load_config()

    def compute(self, value: int) -> int:   # para métodos con args, mejor caché externo
        return value * 2

cached_property almacena el resultado en __dict__ de la instancia, así que vive y muere con el objeto sin el riesgo de retener referencias indefinidamente.

Error: en un decorador con estado mutable compartido entre llamadas, no proteger ese estado con un lock hace que código concurrente produzca resultados incorrectos.

# ❌ Wrong
def count_calls(func):
    count_calls.calls = 0               # estado global sin protección
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        count_calls.calls += 1          # race condition en entornos multi-hilo
        return func(*args, **kwargs)
    return wrapper

# ✅ Right
import threading

def count_calls(func):
    lock = threading.Lock()
    wrapper.calls = 0

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        with lock:
            wrapper.calls += 1
        return func(*args, **kwargs)
    return wrapper

El lock se crea una vez en tiempo de decoración y queda capturado en el closure de wrapper, así cada función decorada tiene su propio lock independiente.

120

Dejar un comentario

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *

Scroll al inicio