Gerenciadores e Iteradores de Contexto Assíncronos
async with entra/sai de configurações e desmontagens que podem ser aguardadas. async for consome iteradores assíncronos que await entre itens - essencial para streaming de corpos HTTP, cursores de banco de dados e frames de websocket.
Receita
from contextlib import asynccontextmanager
@asynccontextmanager
async def session():
print("open")
try:
yield {"ok": True}
finally:
print("close")Quando usar isso:
- Escopos de conexão/sessão de banco de dados
- Respostas de streaming
httpx - Loops de mensagens de Websocket
- Travas distribuídas com aquisição assíncrona
- Streams assíncronos semelhantes a arquivos (aiofiles)
Exemplo de Trabalho
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def managed_connection(name: str):
print("connect", name)
try:
yield name
finally:
print("disconnect", 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("using", conn)
async for value in Countdown(3):
print("tick", value)
asyncio.run(main())O que isso demonstra:
@asynccontextmanagerespelhacontextmanagersíncrono com awaitasync withchama awaitables__aenter__/__aexit____aiter__/__anext__alimentamasync forStopAsyncIterationencerra a iteração assíncrona
Mergulho Profundo
Resumo do Protocolo
| Protocolo | Métodos |
|---|---|
| Gerenciador de contexto assíncrono | __aenter__, __aexit__ |
| Iterador assíncrono | __aiter__, __anext__ |
| Gerador assíncrono | async def com yield |
Padrão de Streaming
async with client.stream("GET", url) as resp:
async for chunk in resp.aiter_bytes():
process(chunk)Armadilhas
- Usando
withsíncrono em um gerenciador assíncrono - não aguarda a desmontagem. Correção: sempreasync with. - Bloqueando em
__aexit__- atrasa o cancelamento. Correção: aguarde caminhos de fechamento não bloqueantes. - Gerador assíncrono não fechado -
aclose()não chamado embreak. Correção: padrãoasync with async_generator()outry/finally. - Misturando
forsíncrono em iterador assíncrono - TypeError. Correção: apenasasync for. async forinfinito sem cancelamento - a tarefa nunca termina. Correção: timeout, evento de desconexão, cancelamento de TaskGroup.
Alternativas
| Alternativa | Usar Quando | Não Usar Quando |
|---|---|---|
try/finally manual com await | Scripts pequenos | Padrão de recurso reutilizável |
Iterador síncrono + to_thread | Stream síncrono legado | Stream assíncrono nativo existe |
| Streaming de callback | SDK antigo | Cliente assíncrono moderno disponível |
FAQs
asynccontextmanager vs classe?
Decorador para yield/finally simples; classe para estado complexo na entrada/saída.
Pode __aexit__ suprimir erros?
Retorne True para suprimir - a mesma semântica dos gerenciadores síncronos.
async for vs while receive?
async for idiomático em protocolos; await recv() manual é bom para sockets com lógica personalizada.
Sair de async for?
break aciona aclose() em geradores assíncronos - verifique se o gerador lida com GeneratorExit.
Múltiplos async with?
async with A(), B(): ou AsyncExitStack de contextlib (suporte assíncrono).
Padrão aiofiles?
async with aiofiles.open(...) as f: await f.read() - E/S de arquivo assíncrona.
Sessão assíncrona SQLAlchemy?
Escopo de transação async with session.begin(): - veja a página de bancos de dados assíncronos.
Testando iteradores assíncronos?
Execute __anext__ até StopAsyncIteration ou colete via async for em pytest-asyncio.
Limpeza de gerador assíncrono?
Use try/finally dentro do gerador; consumidores devem usar async with aclosing(gen()) (3.10+).
Loop de websocket?
async for message in ws: até que a conexão seja fechada - lide com ConnectionClosed graciosamente.
Relacionado
- Gerenciadores de Contexto como Padrão - versão síncrona
- HTTP Assíncrono com httpx/aiohttp - corpos de streaming
- Bancos de Dados Assíncronos - escopos de sessão
- Noções Básicas de Asyncio - introdução ao await
Versões da Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 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+ e uv 0.6+.