Context manager y la sentencia with en Python

1. Descripción general

Abrir un archivo, conectarse a una base de datos o mantener un bloqueo requieren un paso de limpieza al final. Este paso debe ejecutarse incluso si el código intermedio lanza una excepción. Python resuelve esto con context manager y la sentencia with.

Este artículo presenta cómo funciona with bajo el capó y cómo escribir tus propios context managers usando clases o contextlib. También aprenderás cómo manejar excepciones en __exit__, cómo hacer pruebas de context managers y los errores comunes al usarlos.

2. El problema: limpiar recursos con try/finally

Veamos cómo leer un archivo sin usar with:

f = open("orders.csv", encoding="utf-8")
try:
    header = f.readline()
    print(header)
finally:
    f.close()

Esta forma es correcta pero verbosa. También es fácil de olvidar. Si colocas open() dentro del try, y open() falla, entonces f no se asignará y el bloque finally lanzará NameError. La sentencia with encapsula todo el patrón anterior:

with open("orders.csv", encoding="utf-8") as f:
    header = f.readline()
    print(header)

print(f.closed)

order_id,customer,amount

True

Cuando salimos del bloque with, el archivo ha sido cerrado. Esto es cierto tanto si salimos normalmente como si hay una excepción.

3. Protocolo de context manager

Cualquier objeto con dos métodos __enter__ y __exit__ se puede usar con with. Conceptualmente, la sentencia:

with EXPR as target:
    BODY

es equivalente al siguiente código:

manager = EXPR
target = manager.__enter__()
try:
    BODY
except BaseException as e:
    if not manager.__exit__(type(e), e, e.__traceback__):
        raise
else:
    manager.__exit__(None, None, None)

Hay tres puntos a recordar:

• El valor asignado a target es el resultado de __enter__(), no la propia EXPR.
• __exit__ siempre se llama. Cuando hay una excepción, recibe (exc_type, exc_value, traceback). Cuando no hay error, los tres parámetros son None.
• Si __exit__ devuelve un valor truthy, la excepción es tragada. De lo contrario, la excepción se lanza nuevamente.

3.1. Escribir un context manager con una clase

El primer ejemplo es un Timer que mide el tiempo de ejecución de un bloque de código:

import time


class Timer:
    def __init__(self, label: str):
        self.label = label
        self.elapsed: float | None = None

    def __enter__(self):
        self._start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc, tb):
        self.elapsed = time.perf_counter() - self._start
        print(f"[{self.label}] {self.elapsed:.3f}s")
        return False


with Timer("calcular_total_ordenes") as t:
    total = sum(range(5_000_000))

print(t.elapsed > 0)

[calcular_total_ordenes] 0.081s
True

Explicación de cada parte:

• __enter__ registra el momento de inicio y devuelve self. De esta forma, t es el objeto Timer.
• __exit__ calcula el tiempo, imprime en pantalla y devuelve False. Las excepciones si las hay no serán atrapadas.
• La variable t sigue siendo utilizable después del bloque with. La razón es que with no crea un nuevo alcance.

3.2. Manejar excepciones en __exit__

El context manager es más efectivo cuando necesitas reaccionar diferentemente según si el bloque de código es exitoso o falla. El siguiente ejemplo es una transacción simple en un libro mayor (ledger):

class Transaction:
    def __init__(self, ledger: list):
        self.ledger = ledger
        self.pending: list = []

    def __enter__(self):
        return self.pending

    def __exit__(self, exc_type, exc, tb):
        if exc_type is None:
            self.ledger.extend(self.pending)
            print(f"commit {len(self.pending)} registros")
        else:
            print(f"rollback debido a {exc_type.__name__}: {exc}")
        return False


ledger = []

with Transaction(ledger) as tx:
    tx.append({"order_id": 1, "amount": 250_000})
    tx.append({"order_id": 2, "amount": 120_000})

try:
    with Transaction(ledger) as tx:
        tx.append({"order_id": 3, "amount": 90_000})
        raise ValueError("saldo insuficiente")
except ValueError:
    pass

print(len(ledger))

commit 2 registros
rollback debido a ValueError: saldo insuficiente
2

En el segundo bloque with, la orden número 3 no se registra en ledger. Dado que __exit__ devuelve False, ValueError sigue siendo lanzada y la persona que llama decide cómo manejarla.

Nota: Solo devuelve True desde __exit__ cuando intencionalmente quieres tragarte la excepción. Un método __exit__ que accidentalmente return self o devuelve otro objeto truthy hará que el error desaparezca sin dejar rastro alguno.

4. Escribir más compacto con @contextmanager

Escribir una clase completa solo para un paso de setup y teardown es bastante largo. El módulo contextlib proporciona el decorador @contextmanager para convertir un generador en un context manager:

from contextlib import contextmanager


class PaymentService:
    def connect(self):
        print("abriendo conexión de puerta de pago")

    def charge(self, customer: str, amount: int):
        print(f"deduciendo {amount} de {customer}")

    def disconnect(self):
        print("cerrando conexión")


@contextmanager
def payment_session(service: PaymentService):
    service.connect()
    try:
        yield service
    finally:
        service.disconnect()


with payment_session(PaymentService()) as payment:
    payment.charge("Lan", 300_000)

abriendo conexión de puerta de pago
deduciendo 300000 de Lan
cerrando conexión

Cómo el decorador mapea el generador al protocolo de context manager:

• El código antes de yield juega el rol de __enter__.
• El valor yield es el valor asignado a la variable después de as.
• El código después de yield juega el rol de __exit__. Si el bloque with lanza una excepción, esa excepción se lanza nuevamente en la posición yield.

Debido a que la excepción aparece justo en yield, el bloque try/finally alrededor de yield es obligatorio. Sin él, disconnect() no se ejecutará cuando haya un error.

Para manejar tipos específicos de errores por separado, usas except además:

@contextmanager
def safe_charge(order_id: int):
    try:
        yield
    except ConnectionError as e:
        print(f"orden {order_id}: error de red, se reintentará después ({e})")


with safe_charge(42):
    raise ConnectionError("timeout")

print("continuando el procesamiento")

orden 42: error de red, se reintentará después (timeout)
continuando el procesamiento

El generador atrapa el error pero no lo lanza nuevamente, por lo que la excepción es tragada. Este resultado es equivalente a que __exit__ devuelva True.

5. Utilidades disponibles en contextlib

5.1. suppress: ignorar un tipo específico de excepción

import os
from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove("cache_ordenes.tmp")

print("hecho")

hecho

El fragmento anterior es más corto y expresa la intención más claramente que try: ... except FileNotFoundError: pass.

5.2. ExitStack: número desconocido de recursos

Supongamos que necesitas fusionar múltiples archivos de facturas, pero el número de archivos solo se sabe en tiempo de ejecución:

from contextlib import ExitStack
from pathlib import Path


def merge_invoices(paths: list[Path], output: Path) -> None:
    with ExitStack() as stack:
        files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
        with open(output, "w", encoding="utf-8") as out:
            for f in files:
                out.write(f.read())


for i in range(3):
    Path(f"factura_{i}.txt").write_text(f"factura {i}\n", encoding="utf-8")

merge_invoices([Path(f"factura_{i}.txt") for i in range(3)], Path("todas.txt"))
print(Path("todas.txt").read_text(encoding="utf-8"))

factura 0
factura 1
factura 2

enter_context llama a __enter__ y recuerda el __exit__ correspondiente. Cuando ExitStack termina, los archivos se cierran en orden inverso al que se abrieron. Si el tercer archivo no se abre, los dos primeros se cierran correctamente de todas formas.

5.3. closing y nullcontext

• closing(obj) convierte un objeto que solo tiene un método close() en un context manager.
• nullcontext(value) es un context manager que no hace nada. Es útil cuando un parámetro puede ser un context manager o no:

from contextlib import nullcontext
import threading

def update_stock(product: dict, qty: int, lock: threading.Lock | None = None):
    with lock if lock is not None else nullcontext():
        product["stock"] -= qty

product = {"sku": "AO-01", "stock": 10}
update_stock(product, 3)
update_stock(product, 2, threading.Lock())
print(product["stock"])

5

6. Múltiples context managers en una sentencia with

Puedes declarar múltiples context managers en la misma sentencia with. A partir de Python 3.10, la sintaxis con paréntesis permite un salto de línea limpio:

with (
    open("ordenes.csv", encoding="utf-8") as src,
    open("ordenes_backup.csv", "w", encoding="utf-8") as dst,
):
    dst.write(src.read())

Esta forma de escribir es equivalente a dos bloques with anidados. src se abre primero y se cierra último.

7. Async context manager

Con código asyncio, el protocolo correspondiente es __aenter__ y __aexit__, usado con async with. La versión generador de esto es @asynccontextmanager:

import asyncio
from contextlib import asynccontextmanager


@asynccontextmanager
async def order_queue(name: str):
    print(f"conectando cola {name}")
    await asyncio.sleep(0.01)
    try:
        yield [f"orden-{i}" for i in range(3)]
    finally:
        await asyncio.sleep(0.01)
        print(f"cerrando cola {name}")


async def main():
    async with order_queue("pagos") as messages:
        for m in messages:
            print("procesando", m)

asyncio.run(main())

conectando cola pagos
procesando orden-0
procesando orden-1
procesando orden-2
cerrando cola pagos

Nota: Usar with normal con un async context manager causará un error TypeError (o AttributeError en versiones antiguas de Python). Lo opuesto también es cierto. Estos dos protocolos están separados entre sí.

8. Probar context managers

Los context managers deben ser probados tanto en la rama de éxito como en la de error. Primero instala pytest:

pip install pytest

Coloca la clase Transaction de la sección 3.2 en un archivo transaction.py, luego escribe la prueba:

# test_transaction.py
import pytest
from transaction import Transaction


def test_commit_when_no_error():
    ledger = []
    with Transaction(ledger) as tx:
        tx.append({"order_id": 1, "amount": 100})

    assert ledger == [{"order_id": 1, "amount": 100}]


def test_rollback_and_propagate_error():
    ledger = []
    with pytest.raises(ValueError, match="saldo"):
        with Transaction(ledger) as tx:
            tx.append({"order_id": 2, "amount": 999})
            raise ValueError("saldo insuficiente")

    assert ledger == []

pytest -q

..                                                       [100%]
2 passed in 0.01s

El segundo test verifica dos cosas a la vez: que los datos se hayan hecho rollback y que la excepción no sea tragada. El propio pytest.raises también es un context manager, un ejemplo familiar del mismo tema de este artículo.

9. Errores comunes y cuándo no usar

Olvidar try/finally alrededor de yield. Este es el error más común con @contextmanager:

@contextmanager
def broken_session(service):
    service.connect()
    yield service
    service.disconnect()  # no se ejecuta si el bloque with lanza una excepción

Usar un recurso después de que with termine. La variable aún existe, pero el recurso ha sido liberado:

with open("ordenes.csv", encoding="utf-8") as f:
    pass
f.read()

ValueError: I/O operation on closed file.

yield múltiples veces en el generador. El generador usado con @contextmanager debe yield exactamente una vez. Si yield más de una vez, Python lanza RuntimeError: generator didn't stop.

Reutilizar un context manager creado a partir de @contextmanager. Este objeto solo se puede usar una vez. Llamar a with una segunda vez en el mismo objeto fallará. Cada vez que lo uses, llama a la función nuevamente para crear un nuevo objeto.

Abuso. No debes usar context managers en estos casos:

• Los recursos necesitan vivir más tiempo que un bloque de código, por ejemplo, un connection pool que existe durante todo el ciclo de vida de la aplicación. En ese caso, adminístralo en la capa de startup y shutdown de la aplicación.
• La lógica no tiene un par claro de ‘abrir/cerrar’. Una función normal será más fácil de leer.
• Quieres ignorar todas las excepciones usando suppress(Exception). Esta forma oculta bugs igual que except: pass.

Nota: No confíes en el garbage collector o en __del__ para cerrar recursos. El momento en que se ejecutan no está garantizado, especialmente en implementaciones como PyPy. Usa with.

10. Conclusión

Los context managers encapsulan setup y teardown en un lugar y garantizan que teardown siempre se ejecute incluso cuando hay una excepción. Ya has visto:

• El protocolo __enter__ y __exit__, junto con el significado del valor devuelto por __exit__.
• Cómo escribir de forma compacta usando @contextmanager, con try/finally alrededor de yield siendo obligatorio.
• Las utilidades suppress, ExitStack, closing, nullcontext, junto con la versión async.
• Cómo probar tanto la rama de commit como la rama de rollback usando pytest.

Cuando tu código tenga un par de ‘tomar y devolver’ (archivo, bloqueo, transacción, conexión), considera envolverlo en un context manager.