Decoradores y context managers en Python, medidos

Serie: curso de Python 4/6 (intermedio). Para quien usa @algo y with todos los días pero nunca escribió uno. Al terminar vas a poder escribir decoradores con y sin argumentos, vas a saber exactamente qué arregla functools.wraps y vas a construir context managers que limpian bien incluso cuando el cuerpo falla.

Si las rutas de tu framework dejaron de coincidir después de agregar un decorador de logging, si un wrapper "seguro" convirtió un error de JSON en un crash confuso con NoneType tres llamadas más tarde, o si un bloque with se tragó una excepción que necesitabas ver, este curso es para ti. Pasa mucho en el primer proyecto Flask o FastAPI de un bootcamp, y también en el código heredado que te toca mantener como freelancer. Los tres son errores de una sola línea, y abajo cada uno aparece con su salida real. Todo se ejecutó en Python 3.9 (los tiempos, en una Mac con Apple Silicon, M1 Pro) y el comportamiento se volvió a comprobar en Python 3.11.

Los números principales: lru_cache hizo el fib(30) recursivo unas 32,000 veces más rápido (175.9 ms → 5.5 µs); un decorador que no hace nada casi duplicó el costo de una llamada mínima (89 → 180 ns); y lru_cache redujo a la mitad la profundidad de recursión a la que llegas antes del RecursionError (996 → 497 frames).

Las funciones son objetos

Un decorador solo es posible porque una función es un objeto común y corriente: puedes asignarla a otro nombre, guardarla en un dict, pasarla como argumento, devolverla desde otra función y agregarle atributos.

def shout(text):
    """Return text in upper case."""
    return text.upper() + "!"

yell = shout
print(yell("hi"), yell is shout, yell.__name__)
print(type(shout), shout.__doc__)
handlers = {"loud": shout, "quiet": str.lower}
print(handlers["loud"]("ok"), handlers["quiet"]("OK"))

def make_counter():
    count = 0
    def inc():
        nonlocal count
        count += 1
        return count
    return inc
c = make_counter()
print(c(), c(), c.__closure__[0].cell_contents)

HI! True shout
<class 'function'> Return text in upper case.
OK! ok
1 2 2

Fíjate que yell.__name__ sigue siendo shout: el nombre pertenece al objeto función, no a la variable. La última línea muestra un closure: inc mantiene vivo count en una celda después de que make_counter terminó. El curso 2 trató nonlocal y el ámbito; un decorador es exactamente este patrón, con una función como variable capturada.

Cómo escribir un decorador: medir tiempo

@timed encima de un def significa func = timed(func), y se ejecuta una sola vez, al definir la función. timed recibe la función original y devuelve un reemplazo, normalmente un closure llamado wrapper que hace algo antes y después de llamar a la original.

import functools, time

def timed(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        t0 = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            ms = (time.perf_counter() - t0) * 1000
            print(f"[timed] {func.__name__} {ms:.1f} ms")
    return wrapper

@timed
def build_index(n):
    """Map each word to its positions."""
    index = {}
    for i in range(n):
        index.setdefault(f"w{i % 1000}", []).append(i)
    return index

idx = build_index(200_000)
print(len(idx), build_index.__name__, build_index.__doc__)

# what @timed means, without the syntax:
def build_index2(n): return n
build_index2 = timed(build_index2)
build_index2(1)

[timed] build_index 39.7 ms
1000 build_index Map each word to its positions.
[timed] build_index2 0.0 ms

Hay tres hábitos incorporados: *args, **kwargs para que cualquier firma pase sin problema; return func(...) para no perder el resultado; y try/finally para que el tiempo se imprima incluso cuando la función lanza una excepción.

functools.wraps: qué se rompe sin él

Quita la línea @functools.wraps(func) y el objeto devuelto es wrapper, con los metadatos de wrapper:

import functools, inspect, time

def timed_bad(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

def timed(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@timed_bad
def load_users(limit: int = 10) -> list:
    """Load up to `limit` users."""
    return list(range(limit))

@timed
def load_items(limit: int = 10) -> list:
    """Load up to `limit` items."""
    return list(range(limit))

for f in (load_users, load_items):
    print(f.__name__, "|", f.__doc__, "|", inspect.signature(f), "|", f.__qualname__)

wrapper | None | (*args, **kwargs) | timed_bad.<locals>.wrapper
load_items | Load up to `limit` items. | (limit: int = 10) -> list | load_items

Sin wraps se pierden el nombre, el docstring, la firma y el nombre calificado. help(), los tooltips del IDE, Sphinx, la salida de pytest y tus logs leen todo eso. Y empeora cuando algo usa el nombre como clave, como hacen muchos frameworks web y registros de plugins:

import functools, pickle

def logged(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

ROUTES = {}
def route(func):
    ROUTES[func.__name__] = func
    return func

@route
@logged
def home(): return "home"

@route
@logged
def about(): return "about"

print(ROUTES)
print(ROUTES["wrapper"]())
pickle.dumps(home)

{'wrapper': <function logged.<locals>.wrapper at 0x1051f09d0>}
about
Traceback (most recent call last):
  File "~/demo/nowraps2.py", line 23, in <module>
    pickle.dumps(home)
AttributeError: Can't pickle local object 'logged.<locals>.wrapper'

Entraron dos rutas, salió una y no hubo ningún error: about sobrescribió a home en silencio. La falla de pickle importa en cuanto le pasas la función a multiprocessing (curso 5). Agrega @functools.wraps(func) encima de def wrapper y vuelve a ejecutar, cambiando las dos últimas líneas por print(ROUTES["home"](), ROUTES["about"]()) y print(len(pickle.dumps(home)), "bytes pickled"):

{'home': <function home at 0x1011dc820>, 'about': <function about at 0x1011dc940>}
home about
32 bytes pickled

Las dos rutas sobreviven y pickle funciona, porque pickle guarda las funciones por módulo y nombre calificado, y wraps copia __module__, __name__, __qualname__, __doc__, __annotations__ y __dict__, y además pone en __wrapped__ la función original para que inspect pueda ver a través del wrapper.

Ojo con el orden: los decoradores se aplican de abajo hacia arriba, así que @route recibió lo que devolvió @logged.

Decoradores con argumentos: reintentos

@retry(times=4) es una llamada que tiene que devolver un decorador. Son tres funciones anidadas: la fábrica recibe las opciones, el decorador recibe la función y el wrapper recibe los argumentos de la llamada.

import functools, random, time

def retry(times=3, delay=0.01, exceptions=(Exception,)):
    """Decorator factory: retry(...) returns the real decorator."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    print(f"  {func.__name__}: attempt {attempt} failed: {e!r}")
                    if attempt == times:
                        raise
                    time.sleep(delay * 2 ** (attempt - 1))
        return wrapper
    return decorator

rng = random.Random(3)

@retry(times=4, exceptions=(ConnectionError,))
def fetch(url):
    if rng.random() < 0.6:
        raise ConnectionError("reset by peer")
    return f"200 OK {url}"

print(fetch("https://example.com/api"))

@retry(times=2, exceptions=(ConnectionError,))
def parse(text):
    return int(text)

try:
    parse("12a")
except ValueError as e:
    print("not retried:", e)

@retry
def oops():
    return 1
print(oops)
try:
    oops()
except TypeError as e:
    print("TypeError:", e)

  fetch: attempt 1 failed: ConnectionError('reset by peer')
  fetch: attempt 2 failed: ConnectionError('reset by peer')
  fetch: attempt 3 failed: ConnectionError('reset by peer')
200 OK https://example.com/api
not retried: invalid literal for int() with base 10: '12a'
<function retry.<locals>.decorator at 0x102e4c550>
TypeError: decorator() missing 1 required positional argument: 'func'

Lo que vale la pena copiar: reintenta solo las excepciones que tiene sentido reintentar (un ValueError por una entrada mal formada va a fallar igual siempre), espera con backoff exponencial, y vuelve a lanzar con un raise solo en el último intento para que se conserve el traceback original. El último bloque es el desliz clásico: @retry sin paréntesis le pasa oops como times, así que oops se convierte en el decorator interno, y el error solo aparece cuando alguien la llama. En Python 3.11 el mensaje lo nombra con más claridad: retry.<locals>.decorator() missing 1 required positional argument: 'func'.

Caché: un memoize escrito a mano y después lru_cache

Un decorador de caché es un dict dentro del closure:

import functools

def memoize(func):
    cache = {}
    @functools.wraps(func)
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    wrapper.cache = cache                  # functions can carry attributes
    return wrapper

@memoize
def slow_square(x):
    print(f"  computing {x}")
    return x * x

print(slow_square(4), slow_square(4), slow_square.cache)

  computing 4
16 16 {(4,): 16}

No tiene límite de tamaño, ignora los argumentos por nombre y no lleva estadísticas. functools.lru_cache, de la biblioteca estándar, resuelve todo eso en C. Esta es la medición contra la recursión ingenua, con la caché vacía en cada ejecución:

"""Naive fib vs lru_cache fib (Python 3.9, Apple M1 Pro)."""
import functools, sys, time

def fib_plain(n):
    return n if n < 2 else fib_plain(n - 1) + fib_plain(n - 2)

@functools.lru_cache(maxsize=None)
def fib_cached(n):
    return n if n < 2 else fib_cached(n - 1) + fib_cached(n - 2)

def best(fn, arg, repeat=5, clear=None):
    out = float("inf")
    for _ in range(repeat):
        if clear: clear()                      # cold cache every run
        t0 = time.perf_counter(); fn(arg); out = min(out, time.perf_counter() - t0)
    return out

print("Python", sys.version.split()[0])
for n in (20, 25, 30):
    tp = best(fib_plain, n)
    tc = best(fib_cached, n, clear=fib_cached.cache_clear)
    print(f"fib({n}) = {fib_plain(n):>6}  plain {tp*1000:7.2f} ms  "
          f"lru_cache {tc*1e6:5.1f} us  x{tp/tc:,.0f}")

fib_cached.cache_clear()
fib_cached(30)
print(fib_cached.cache_info())
fib_cached(30)
print(fib_cached.cache_info())

Python 3.9.6
fib(20) =   6765  plain    1.43 ms  lru_cache   3.9 us  x370
fib(25) =  75025  plain   15.78 ms  lru_cache   4.9 us  x3,210
fib(30) = 832040  plain  175.90 ms  lru_cache   5.5 us  x31,740
CacheInfo(hits=28, misses=31, maxsize=None, currsize=31)
CacheInfo(hits=29, misses=31, maxsize=None, currsize=31)

Una segunda ejecución dio 1.44/16.16/175.96 ms y 3.7/4.8/5.5 µs. La versión simple crece unas 11 veces por cada +5 (hace 2,692,537 llamadas para fib(30)); la versión con caché crece en forma lineal. cache_info() es la prueba: 31 misses para los 31 argumentos distintos 0..30, 28 hits para la segunda rama de cada llamada, y la llamada repetida suma un hit más.

Dos trampas de lru_cache

@functools.lru_cache(maxsize=128)
def total(items):
    return sum(items)
print(total((1, 2, 3)))
total([1, 2, 3])

6
Traceback (most recent call last):
  File "~/demo/memo.py", line 24, in <module>
    total([1, 2, 3])
TypeError: unhashable type: 'list'

Los argumentos se convierten en claves de un dict, así que tienen que ser hashables: pasa tuplas o frozensets.

La segunda trampa nos sorprendió. Cada llamada recursiva a través de lru_cache consume dos niveles del límite de recursión en lugar de uno. Medimos el n más profundo que todavía funciona con un simple f(n) = 1 + f(n - 1) y el límite por defecto de 1000:

recursionlimit 1000
plain max depth  996
cached max depth 497

Python 3.11 dio 997 y 498. O sea, una función recursiva memoizada con la caché vacía choca con RecursionError: maximum recursion depth exceeded a la mitad de la profundidad que esperarías. Calienta la caché de abajo hacia arriba (llama a f(200), f(400), …) o reescríbela como bucle.

Context managers: el protocolo de with

with EXPR as x: llama a EXPR.__enter__() y asigna el resultado a x, ejecuta el cuerpo y después llama siempre a __exit__(exc_type, exc, tb), con tres None si el cuerpo terminó bien. Si __exit__ devuelve un valor verdadero, la excepción se suprime.

import contextlib, os, tempfile, time

class Timer:
    def __enter__(self):
        self.t0 = time.perf_counter()
        print("enter")
        return self                       # bound to the name after `as`
    def __exit__(self, exc_type, exc, tb):
        self.elapsed = time.perf_counter() - self.t0
        print(f"exit: exc_type={exc_type.__name__ if exc_type else None}")
        return False                      # do not swallow the exception

with Timer() as t:
    sum(range(100_000))
print("elapsed set:", t.elapsed > 0)

try:
    with Timer():
        1 / 0
except ZeroDivisionError:
    print("ZeroDivisionError still reached the caller")

enter
exit: exc_type=None
elapsed set: True
enter
exit: exc_type=ZeroDivisionError
ZeroDivisionError still reached the caller

En la mayoría de los casos, contextlib.contextmanager convierte un generador en un context manager: el código antes del yield es __enter__, el valor entregado es lo que recibe as, y el código de después es __exit__. El try/finally no es opcional:

@contextlib.contextmanager
def chdir_tmp():
    old = os.getcwd()
    path = tempfile.mkdtemp()
    os.chdir(path)
    try:
        yield path
    finally:                              # without try/finally, an error skips cleanup
        os.chdir(old)
        os.rmdir(path)

Sin él, una excepción en el cuerpo se vuelve a lanzar en el yield, y todo lo que viene después del yield se salta:

@contextlib.contextmanager
def opened_resource(name):
    print(f"acquire {name}")
    yield name
    print(f"release {name}")          # BUG: no try/finally

try:
    with opened_resource("db-conn"):
        raise ValueError("query failed")
except ValueError as e:
    print("caught:", e)

acquire db-conn
caught: query failed

No aparece la línea "release": en código real, eso es una conexión o un lock que queda abierto.

contextlib.suppress es la forma explícita y acotada de ignorar un tipo de excepción, y ExitStack maneja una cantidad de recursos que solo se conoce en tiempo de ejecución, y los cierra en orden inverso:

with contextlib.suppress(FileNotFoundError):
    os.remove("does-not-exist.txt")
print("suppress: continued")

paths = []
for i in range(3):
    fd, p = tempfile.mkstemp(suffix=f"-{i}.txt"); os.close(fd); paths.append(p)
with contextlib.ExitStack() as stack:
    files = [stack.enter_context(open(p, "w")) for p in paths]
    stack.callback(print, "callback: runs last-in, first-out")
    for i, f in enumerate(files):
        f.write(f"part {i}\n")
    print("open:", [f.closed for f in files])
print("closed:", [f.closed for f in files])
for p in paths: os.remove(p)

suppress: continued
open: [False, False, False]
callback: runs last-in, first-out
closed: [True, True, True]

El callback se registró último, así que se ejecutó primero, antes de cerrar los tres archivos. Si abrir el tercer archivo hubiera fallado, los dos primeros se habrían cerrado igual.

Errores comunes: el decorador que se come resultados y errores

Dos bugs que parecen inofensivos en un code review:

import functools, logging
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")

def logged(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logging.info("call %s%r", func.__name__, args)
        func(*args, **kwargs)                       # BUG 1: result is dropped
    return wrapper

def safe(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:                      # BUG 2: swallowed, returns None
            logging.error("%s failed: %s", func.__name__, e)
    return wrapper

@logged
def price_with_tax(yen):
    return round(yen * 1.1)

total = price_with_tax(1000)
print("total:", total)

@safe
def load_config(text):
    import json
    return json.loads(text)

cfg = load_config('{"retries": 3,}')
print(cfg["retries"])

INFO call price_with_tax(1000,)
total: None
ERROR load_config failed: Expecting property name enclosed in double quotes: line 1 column 15 (char 14)
Traceback (most recent call last):
  File "~/demo/deco_bad.py", line 33, in <module>
    print(cfg["retries"])
TypeError: 'NoneType' object is not subscriptable

Qué pasa en realidad:

• return olvidado: ahora todas las funciones decoradas devuelven None. Sin ningún error; total: None sigue de largo hacia el próximo cálculo o hacia una respuesta JSON.
• Excepción tragada: la causa real (una coma de más al final del JSON) se registra una vez donde ocurre, y el crash llega después, en otro lugar, como un TypeError sobre NoneType. En un log con mucho movimiento, nadie conecta las dos líneas.

Solución: siempre return func(*args, **kwargs). En un decorador que atrapa excepciones, o vuelves a lanzarla (except Exception: log(...); raise) o atrapas solo excepciones concretas y esperadas y devuelves un valor que quien llama está preparado para manejar. La misma regla vale para __exit__:

class Swallow:
    def __enter__(self): return self
    def __exit__(self, *exc):
        print("cleanup")
        return True                    # BUG: True means "exception handled"

with Swallow():
    raise KeyError("user_id")
print("the KeyError vanished")

cleanup
the KeyError vanished

Devuelve True desde __exit__ solo a propósito, y solo después de revisar exc_type.

Tabla de decisión

Y un costo para recordar: un decorador que solo pasa la llamada, alrededor de una función de dos argumentos, tardó 178–183 ns por llamada contra 89–90 ns de la función sola (1,000,000 de llamadas, mejor de 5, dos ejecuciones). No es nada al lado de una llamada de red, y se nota dentro de un bucle ajustado sobre millones de elementos.

Límites honestos y notas de versión

• Tiempos: Python 3.9.6 en una Mac con Apple Silicon (M1 Pro), time.perf_counter, mejor de 5, cada ejecución dos veces; semillas fijas (random.Random(3) para la demo de reintentos). Python 3.11 se usó solo para comprobar comportamiento y mensajes, nunca para medir tiempos.
• functools.cache (un lru_cache sin límite) existe desde 3.9, y @lru_cache sin paréntesis desde 3.8; ambos confirmados en 3.9. contextlib.chdir existe desde 3.11 (confirmamos que falta en 3.9 y está en 3.11), por eso el ejemplo de arriba escribe el suyo.
• Las cifras de profundidad de recursión (996/497 en 3.9, 997/498 en 3.11) son propias de estas versiones. Python 3.12 rehízo la forma de limitar la recursión a través de código C, así que ahí los números pueden ser otros; 3.12+ no se ejecutó.
• Los tracebacks de Python 3.11 agregan marcadores ^^^^ debajo de la expresión que falla; aparte de eso, los mensajes fueron iguales.

Navegación de la serie: anterior: Curso de Python 3: iteradores, generadores y memoria · siguiente: Curso de Python 5: hilos, procesos, asyncio y el GIL. Si vienes de C#, los decoradores cumplen el papel que allí se reparten los delegados (en inglés) y los atributos, y with es el using de C#; mira también el manejo de excepciones en C# (en inglés) para el problema de la excepción tragada del lado de .NET.