`contextlib.contextmanager`: context managers con generadores

Cuando necesitas un context manager de un solo uso o quieres encapsular lógica de setup/teardown sin escribir una clase completa, @contextmanager es la herramienta exacta para ese trabajo.

La idea es simple: tomas una función generadora con exactamente un yield y el decorador la transforma en un context manager válido. Todo lo que escribes antes del yield se ejecuta como __enter__; todo lo que escribes después se ejecuta como __exit__. El valor que produces con yield es lo que recibe la cláusula as.

¿Por qué funciona así? Porque @contextmanager envuelve tu generador en un objeto _GeneratorContextManager que implementa el protocolo de context manager por ti. Cuando Python ejecuta with tu_funcion() as x:, llama a __enter__, que hace next() sobre el generador hasta el yield. Al salir del bloque with, llama a __exit__, que reanuda el generador desde ese punto. Es una delegación elegante del flujo de control.

Lo usas cuando el setup y teardown son conceptualmente una sola pieza de lógica que no justifica una clase entera, o cuando escribes fixtures, patches temporales o wrappers de medición que viven en un único lugar. Escribir __enter__ y __exit__ directamente implica mantener estado entre los dos métodos (normalmente en self), separar código que lógicamente va junto, y recordar la firma exacta de __exit__. Con @contextmanager esos problemas desaparecen.

El punto crítico que no puedes ignorar: si el bloque with lanza una excepción y no la capturas dentro de tu generador, la excepción se re-lanza en el punto del yield y el código posterior no se ejecuta. Esto significa que si omites el try/finally, tu teardown se saltará silenciosamente. La solución es siempre envolver el yield en try/finally cuando tengas recursos que liberar.

import time
import contextlib
from typing import Iterator


# ── 1. Medición de tiempo de un bloque ───────────────────────────────────────

@contextlib.contextmanager
def timer(label: str) -> Iterator[None]:
    start = time.perf_counter()
    try:
        yield  # el bloque with se ejecuta aquí
    finally:
        # finally garantiza que esto corre incluso si el bloque lanza
        elapsed = time.perf_counter() - start
        print(f"[{label}] {elapsed:.4f}s")


# ── 2. Patch temporal de configuración ───────────────────────────────────────

@contextlib.contextmanager
def override_config(config: dict, **overrides) -> Iterator[dict]:
    original = {k: config[k] for k in overrides if k in config}
    missing = {k for k in overrides if k not in config}
    config.update(overrides)
    try:
        yield config  # exponemos el dict modificado al bloque with
    finally:
        # restauramos exactamente los valores anteriores
        for k in overrides:
            if k in missing:
                config.pop(k, None)
            else:
                config[k] = original[k]


# ── 3. Transacción de test que se revierte ────────────────────────────────────

class FakeDB:
    """Simula una base de datos con soporte de savepoints."""

    def __init__(self) -> None:
        self.rows: list[str] = ["alice", "bob"]
        self._snapshot: list[str] | None = None

    def savepoint(self) -> None:
        self._snapshot = self.rows.copy()

    def rollback(self) -> None:
        if self._snapshot is not None:
            self.rows = self._snapshot


@contextlib.contextmanager
def revert_after(db: FakeDB) -> Iterator[FakeDB]:
    db.savepoint()
    try:
        yield db
    except Exception:
        raise  # re-lanzamos para no tragarnos errores inesperados
    finally:
        db.rollback()  # siempre revertimos, con o sin excepción


# ── Demostración ──────────────────────────────────────────────────────────────

def main() -> None:
    # Timer
    with timer("fibonacci"):
        result = sum(range(100_000))
    print(f"resultado: {result}")

    # Patch de configuración
    settings = {"debug": False, "timeout": 30}
    with override_config(settings, debug=True, timeout=5) as cfg:
        print(f"dentro del bloque: {cfg}")
    print(f"fuera del bloque:  {settings}")  # valores originales restaurados

    # Transacción revertible
    db = FakeDB()
    with revert_after(db) as session:
        session.rows.append("charlie")
        print(f"durante el test: {session.rows}")
    print(f"después del test: {db.rows}")  # charlie no está


if __name__ == "__main__":
    main()

Qué hace cada decisión

El timer es el ejemplo más puro: no hay valor que exponer al bloque with, así que el yield va solo. El try/finally no es opcional aquí —si el código que mides lanza, aún quieres ver cuánto tardó antes de explotar.

En override_config, el yield config expone el objeto mutado. Fíjate que guardamos solo las claves que vamos a modificar y distinguimos entre claves que existían y claves nuevas que añadimos; sin esa distinción, el teardown podría dejar en el diccionario claves que no estaban antes.

revert_after muestra un patrón habitual en tests: el except Exception: raise es deliberado. No queremos silenciar errores; solo queremos asegurarnos de que el rollback ocurra sin importar qué. Podrías omitir el except completamente y dejar solo el finally, pero hacerlo explícito comunica la intención a quien lea el código.

Una cosa que sorprende a gente que viene de escribir __exit__ a mano: si quisieras inspeccionar o suprimir la excepción (el equivalente a retornar True desde __exit__), tendrías que capturarla dentro del try alrededor del yield. Si la capturas y no la re-lanzas, @contextmanager interpreta eso como “excepción suprimida”. Si en cambio el generador termina sin recibir la excepción (porque llegó al final normalmente), también la suprime. El único comportamiento que re-lanza es que la excepción salga del generador sin capturarse.

Errores que debes conocer

Error: olvidar el try/finally cuando hay recursos que liberar, asumiendo que el código después del yield siempre se ejecuta.

# ❌ Wrong
@contextlib.contextmanager
def open_connection(url: str):
    conn = connect(url)
    yield conn
    conn.close()  # no se ejecuta si el bloque with lanza

# ✅ Right
@contextlib.contextmanager
def open_connection(url: str):
    conn = connect(url)
    try:
        yield conn
    finally:
        conn.close()  # se ejecuta siempre

El finally garantiza que close() se llama incluso si el bloque with lanza cualquier excepción.

Error: usar más de un yield dentro del generador, lo que hace que @contextmanager lance RuntimeError al salir del bloque.

# ❌ Wrong
@contextlib.contextmanager
def bad_manager():
    yield "first"
    yield "second"  # RuntimeError: generator didn't stop after throw()

# ✅ Right
@contextlib.contextmanager
def good_manager():
    yield "only one"

El decorador espera exactamente un punto de suspensión; dos yield rompen el contrato del protocolo.

Error: intentar usar @contextmanager en una función que devuelve un generador anidado sin yield directo, lo que produce un context manager que nunca entrega el valor esperado.

# ❌ Wrong
@contextlib.contextmanager
def bad():
    def inner():
        yield 42  # este yield es del generador interno, no del externo
    inner()  # bad() no tiene yield propio → RuntimeError al entrar

# ✅ Right
@contextlib.contextmanager
def good():
    yield 42  # el yield debe estar en el cuerpo directo de la función decorada

El yield tiene que pertenecer a la función decorada, no a una función anidada dentro de ella.

122

Dejar un comentario

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

Scroll al inicio