Gestores y Iteradores de Contexto Asíncronos

async with entra y sale de configuraciones y limpiezas que se pueden esperar. async for consume iteradores asíncronos que hacen await entre elementos, lo cual es esencial para el streaming de cuerpos HTTP, cursores de bases de datos y tramas de websocket.

Receta

from contextlib import asynccontextmanager
 
@asynccontextmanager
async def session():
    print("abrir")
    try:
        yield {"ok": True}
    finally:
        print("cerrar")

Cuándo usar esto:

• Ámbitos de conexión/sesión de base de datos
• Respuestas de streaming de httpx
• Bucles de mensajes de Websocket
• Bloqueos distribuidos con adquisición asíncrona
• Flujos asíncronos similares a archivos (aiofiles)

Ejemplo de Trabajo

import asyncio
from contextlib import asynccontextmanager
 
@asynccontextmanager
async def managed_connection(name: str):
    print("conectar", name)
    try:
        yield name
    finally:
        print("desconectar", name)
 
class Countdown:
    def __init__(self, start: int) -> None:
        self._n = start
 
    def __aiter__(self):
        return self
 
    async def __anext__(self) -> int:
        if self._n <= 0:
            raise StopAsyncIteration
        await asyncio.sleep(0.01)
        self._n -= 1
        return self._n + 1
 
async def main() -> None:
    async with managed_connection("db") as conn:
        print("usando", conn)
    async for value in Countdown(3):
        print("tick", value)
 
asyncio.run(main())

Lo que esto demuestra:

• @asynccontextmanager refleja contextmanager síncrono con await
• async with llama a los awaitables __aenter__/__aexit__
• __aiter__/__anext__ potencian async for
• StopAsyncIteration finaliza la iteración asíncrona

Inmersión Profunda

Resumen de Protocolos

Patrón de Streaming

async with client.stream("GET", url) as resp:
    async for chunk in resp.aiter_bytes():
        process(chunk)

Trampas Comunes

• Usar with síncrono en un gestor asíncrono - no espera la limpieza. Solución: siempre usar async with.
• Bloqueo en __aexit__ - retrasa la cancelación. Solución: esperar rutas de cierre no bloqueantes.
• Generador asíncrono no cerrado - aclose() no se llama en break. Solución: patrón async with async_generator() o try/finally.
• Mezclar for síncrono en un iterador asíncrono - TypeError. Solución: solo async for.
• async for infinito sin cancelación - la tarea nunca termina. Solución: tiempo de espera, evento de desconexión, cancelación de TaskGroup.

Alternativas

Preguntas Frecuentes

Relacionado

• Context Managers as a Pattern - versión síncrona
• Async HTTP with httpx/aiohttp - cuerpos de streaming
• Async Databases - ámbitos de sesión
• Asyncio Basics - introducción a await

Versiones de la pila: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, y uv 0.6+.