Async Context Managers & Iterators
async with enters/exits awaitable setup and teardown. async for consumes async iterators that await between items - essential for streaming HTTP bodies, DB cursors, and websocket frames.
Recipe
from contextlib import asynccontextmanager
@asynccontextmanager
async def session():
print("open")
try:
yield {"ok": True}
finally:
print("close")When to reach for this:
- DB connection/session scopes
httpxstreaming responses- Websocket message loops
- Distributed locks with async acquire
- File-like async streams (aiofiles)
Working Example
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())What this demonstrates:
@asynccontextmanagermirrors synccontextmanagerwith awaitasync withcalls__aenter__/__aexit__awaitables__aiter__/__anext__powerasync forStopAsyncIterationends async iteration
Deep Dive
Protocol Summary
| Protocol | Methods |
|---|---|
| Async context manager | __aenter__, __aexit__ |
| Async iterator | __aiter__, __anext__ |
| Async generator | async def with yield |
Streaming Pattern
async with client.stream("GET", url) as resp:
async for chunk in resp.aiter_bytes():
process(chunk)Gotchas
- Using sync
withon async manager - does not await teardown. Fix: alwaysasync with. - Blocking in
__aexit__- delays cancel. Fix: await non-blocking close paths. - Async generator not closed -
aclose()not called on break. Fix:async with async_generator()pattern or try/finally. - Mixing sync for on async iter - TypeError. Fix:
async foronly. - Infinite async for without cancel - task never ends. Fix: timeout, disconnect event, TaskGroup cancel.
Alternatives
| Alternative | Use When | Don't Use When |
|---|---|---|
| Manual try/finally await | Tiny scripts | Reusable resource pattern |
| Sync iterator + to_thread | Legacy sync stream | Native async stream exists |
| Callback streaming | Old SDK | Modern async client available |
FAQs
asynccontextmanager vs class?
Decorator for simple yield/finally; class when complex state on enter/exit.
Can __aexit__ suppress errors?
Return True to suppress - same semantics as sync managers.
async for vs while receive?
async for idiomatic on protocols; manual await recv() fine for sockets with custom logic.
Breaking out of async for?
break triggers aclose() on async generators - verify generator handles GeneratorExit.
Multiple async with?
async with A(), B(): or AsyncExitStack from contextlib (async support).
aiofiles pattern?
async with aiofiles.open(...) as f: await f.read() - async file I/O.
SQLAlchemy async session?
async with session.begin(): transaction scope - see async databases page.
Testing async iterators?
Drive __anext__ until StopAsyncIteration or collect via async for in pytest-asyncio.
async generator cleanup?
Use try/finally inside generator; consumers should async with aclosing(gen()) (3.10+).
websocket loop?
async for message in ws: until connection closed - handle ConnectionClosed gracefully.
Related
- Context Managers as a Pattern - sync version
- Async HTTP with httpx/aiohttp - streaming bodies
- Async Databases - session scopes
- Asyncio Basics - await primer
Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 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+, and uv 0.6+.