El módulo contextlib ofrece herramientas para crear y combinar context managers. Controlan la entrada y salida de un bloque with, garantizando cleanup de archivos, sockets, locks, transacciones, directorios temporales y otros recursos incluso cuando ocurre una excepción. El módulo reduce clases repetitivas y hace explícitos el lifecycle y el manejo de errores.
Un context manager no sirve únicamente para cerrar objetos. Puede configurar estado temporal, registrar métricas, redirigir streams, iniciar y finalizar transacciones o componer una cantidad dinámica de recursos. La regla central es que cada adquisición exitosa tenga una liberación correspondiente y predecible.
El protocolo de contexto
Un context manager implementa __enter__() y __exit__(). El valor devuelto por __enter__() se asigna al objetivo de as. __exit__() recibe información de la excepción cuando el bloque falla.
class Recurso:
def __enter__(self):
self.abrir()
return self
def __exit__(self, tipo, valor, traceback):
self.cerrar()
return False
Devolver un valor verdadero desde __exit__() suprime la excepción. Hazlo solo cuando el error haya sido tratado realmente.
contextmanager
El decorator @contextmanager convierte una función generator en context manager. El código anterior a yield se ejecuta al entrar; el cleanup dentro de finally, al salir.
from contextlib import contextmanager
@contextmanager
def conexion_temporal():
conexion = abrir_conexion()
try:
yield conexion
finally:
conexion.close()
with conexion_temporal() as conexion:
conexion.ejecutar()
El generator debe producir exactamente una vez. No alcanzar el yield o producir varias veces rompe el protocolo.
Coloca cleanup en finally
Sin finally, una excepción lanzada por el cuerpo puede saltar la liberación.
@contextmanager
def archivo_bloqueado(ruta):
archivo = open(ruta, "a+")
adquirir_lock(archivo)
try:
yield archivo
finally:
liberar_lock(archivo)
archivo.close()
Si la adquisición puede fallar a mitad, libera solamente recursos realmente adquiridos.
Excepciones dentro del generator
Cuando el cuerpo lanza una excepción, se inyecta en el punto de yield. El generator puede registrarla, convertirla o tratarla.
@contextmanager
def registrar_fallos(logger):
try:
yield
except Exception:
logger.exception("fallo en el bloque")
raise
Vuelve a lanzar cuando el problema no haya sido resuelto. Registrar y continuar puede ocultar estado corrupto.
ContextDecorator
Los context managers basados en ContextDecorator también pueden decorar funciones.
from contextlib import ContextDecorator
class Cronometro(ContextDecorator):
def __enter__(self):
self.inicio = ahora()
return self
def __exit__(self, *exc):
registrar_duracion(ahora() - self.inicio)
return False
@Cronometro()
def procesar():
ejecutar_tarea()
El objeto debe soportar uso repetido cuando la función decorada pueda llamarse varias veces.
closing
closing(objeto) llama a close() al salir. Es útil para objetos legacy que tienen cierre, pero no implementan el protocolo.
from contextlib import closing
with closing(abrir_recurso_legacy()) as recurso:
recurso.usar()
No envuelvas un objeto que ya soporta with sin necesidad. Su manager nativo puede ejecutar pasos adicionales.
aclosing
aclosing() es la versión asíncrona para objetos con aclose(), especialmente generators async.
from contextlib import aclosing
async with aclosing(stream_asincrono()) as stream:
async for item in stream:
if item.listo:
break
El cleanup ocurre en el mismo contexto asíncrono, conservando context variables, excepciones y lifecycle de la tarea.
asynccontextmanager
@asynccontextmanager crea context managers async a partir de async generators.
from contextlib import asynccontextmanager
@asynccontextmanager
async def cliente_api():
cliente = await crear_cliente()
try:
yield cliente
finally:
await cliente.aclose()
Úsalo con async with. El cleanup puede esperar I/O, pero necesita timeouts y cancelación correctos.
Managers reutilizables y reentrantes
Algunos managers son single-use, otros reutilizables y otros reentrantes. Son propiedades diferentes.
Un manager basado en generator crea una instancia nueva cada vez que llamas la función decorada; usa with recurso(): y no almacenes una instancia para múltiples entradas.
nullcontext
nullcontext() no realiza cleanup y devuelve un valor opcional. Simplifica caminos donde el recurso quizá ya esté abierto.
from contextlib import nullcontext
contexto = open(ruta) if ruta else nullcontext(stream_existente)
with contexto as stream:
procesar(stream)
Evita duplicar el cuerpo del with en dos branches.
suppress
suppress(*excepciones) ignora tipos seleccionados.
from contextlib import suppress
with suppress(FileNotFoundError):
ruta.unlink()
Úsalo solo cuando la excepción represente un resultado esperado y aceptable. No suprimas Exception ampliamente.
Suprimir no es registrar
Si un error requiere auditoría, retry, métrica o decisión visible, un try/except explícito es más claro. suppress comunica que la falta de efecto es aceptable y silenciosa.
redirect_stdout
redirect_stdout(destino) reemplaza temporalmente sys.stdout.
from contextlib import redirect_stdout
from io import StringIO
buffer = StringIO()
with redirect_stdout(buffer):
funcion_que_imprime()
texto = buffer.getvalue()
El cambio es global al proceso y afecta otras threads. Úsalo principalmente en scripts, tests controlados y herramientas single-thread.
redirect_stderr
redirect_stderr() hace lo mismo con sys.stderr. No captura escrituras directas a descriptores nativos, subprocesses independientes o logging configurado en otro destino.
Para procesos hijos, usa las opciones de captura de subprocess.
chdir temporal
chdir(ruta) cambia el directorio actual durante el bloque y lo restaura al salir.
from contextlib import chdir
with chdir("proyecto"):
ejecutar_build()
El directorio actual es estado global. No uses este patrón en programas con threads o tareas concurrentes que dependan de paths relativos.
ExitStack
ExitStack compone una cantidad dinámica de context managers y callbacks.
from contextlib import ExitStack
with ExitStack() as stack:
archivos = [
stack.enter_context(open(ruta, encoding="utf-8"))
for ruta in rutas
]
combinar(archivos)
Si abrir el tercer archivo falla, los anteriores se cierran automáticamente.
Orden LIFO
ExitStack ejecuta callbacks en orden inverso a la adquisición. Esto encaja con recursos dependientes: el más reciente se libera primero.
Registra cleanup inmediatamente después de cada adquisición para no crear una ventana de fuga.
callback
stack.callback(funcion, *args, **kwargs) registra una llamada de cleanup que no recibe información de la excepción.
with ExitStack() as stack:
directorio = crear_directorio_temporal()
stack.callback(eliminar_arbol, directorio)
ejecutar(directorio)
Haz callbacks idempotentes cuando sea posible, porque el cleanup puede encontrar estado parcial.
push
push() registra la parte de salida de un context manager o una función compatible con __exit__. A diferencia de callback, puede observar y suprimir excepciones.
Usa esa capacidad con cuidado, porque cambia lo que ven callbacks exteriores y callers.
enter_context
enter_context(cm) llama a __enter__() y registra __exit__(). Devuelve el valor que recibiría un objetivo as.
Así resulta sencillo construir recursos definidos en runtime.
pop_all
pop_all() transfiere callbacks a otra stack sin ejecutarlos. Sirve para adquisición “todo o nada”.
stack = ExitStack()
try:
recursos = [stack.enter_context(abrir(x)) for x in items]
except Exception:
stack.close()
raise
else:
stack_final = stack.pop_all()
Después de transferir, el nuevo owner debe cerrar la stack.
AsyncExitStack
AsyncExitStack combina context managers síncronos, async y callbacks de cleanup async.
from contextlib import AsyncExitStack
async with AsyncExitStack() as stack:
clientes = [
await stack.enter_async_context(crear_cliente(url))
for url in urls
]
await consultar(clientes)
Es especialmente útil cuando la cantidad de conexiones asíncronas es dinámica.
push_async_callback
Los callbacks asíncronos pueden esperar flush, shutdown o liberación. También ejecutan en orden inverso.
Aplica deadlines, porque un cleanup trabado puede impedir el cierre de la aplicación.
Transacciones
Un manager de transacción puede hacer commit tras completar normalmente y rollback ante una excepción.
@contextmanager
def transaccion(conexion):
try:
yield conexion
except Exception:
conexion.rollback()
raise
else:
conexion.commit()
No ocultes el fallo original después del rollback sin otro canal de estado explícito.
Adquisición parcial
Cuando setup tiene varias etapas, usa una ExitStack interna para registrar cada cleanup. Transfiere la stack solo cuando todas las etapas hayan tenido éxito.
Este patrón sustituye flags booleanas y finally anidados.
Excepciones durante cleanup
Una excepción de cleanup puede sustituir o encadenarse con el error original. Usa tipos específicos, logging y chaining explícito para conservar diagnóstico.
No ignores un commit, flush o close fallido cuando garantiza durabilidad.
Varios managers en un with
Para una cantidad fija, un único with con varios managers es más simple.
with abrir_a() as a, abrir_b() as b:
usar(a, b)
Usa ExitStack cuando cantidad o tipos sean dinámicos.
Context managers y sockets
Los sockets ya soportan el protocolo y se cierran al salir.
import socket
with socket.create_connection((host, puerto), timeout=5) as sock:
sock.sendall(datos)
Consulta socket en Python para timeouts, framing y shutdown.
Context managers y e-mail
Los clientes smtplib.SMTP también soportan with, garantizando cierre ordenado. Consulta smtplib en Python.
Locks
Locks de threading y multiprocessing funcionan como context managers. El bloque reduce el riesgo de olvidar liberar.
with lock:
actualizar_estado()
Todavía debes evitar deadlock, mantener corta la sección crítica y adquirir varios locks en orden consistente.
Context variables
Un manager puede configurar temporalmente una ContextVar y restaurar su token al salir.
@contextmanager
def contexto_request(valor):
token = request_id.set(valor)
try:
yield
finally:
request_id.reset(token)
Es útil para logging y tracing, también en código async.
Métricas
Un manager puede medir duración y registrar éxito o fallo al salir mientras preserva la excepción.
No permitas que una falla del backend de observabilidad oculte el error principal.
Tipado
Usa typing.ContextManager, AsyncContextManager o protocolos estructurales para declarar APIs. Anota el tipo producido por yield.
Una firma clara distingue el objeto manager del recurso que devuelve.
Pruebas
Prueba salida normal, excepción en el cuerpo, fallo de adquisición, fallo de cleanup, uso repetido, cancelación async y adquisición parcial. Verifica el orden de liberación.
Los mocks deben confirmar que close, rollback y callbacks ocurren exactamente cuando corresponde.
Errores comunes
Los fallos frecuentes son olvidar finally, producir varias veces en @contextmanager, suprimir excepciones accidentalmente, redirigir stdout en un proceso multithread, cambiar el directorio global concurrentemente, reutilizar un manager single-use, registrar cleanup demasiado tarde y olvidar cerrar una stack transferida.
Conclusión
contextlib hace más seguras y claras la adquisición y liberación. Usa @contextmanager para managers simples, ExitStack para recursos dinámicos, variantes async para cleanup esperable y nullcontext para caminos opcionales.
Mantén ownership explícito y no ocultes errores importantes. Consulta la documentación oficial de contextlib y la referencia del protocolo de context manager.
