Async Context Managers & Iterators

Q: Can __aexit__ suppress errors?

Return True to suppress - same semantics as sync managers.

Q: async for vs while receive?

async for idiomatic on protocols; manual await recv() fine for sockets with custom logic.

Q: Breaking out of async for?

break triggers aclose() on async generators - verify generator handles GeneratorExit.

Q: Multiple async with?

async with A(), B(): or AsyncExitStack from contextlib (async support).

Q: aiofiles pattern?

async with aiofiles.open(...) as f: await f.read() - async file I/O.

Q: SQLAlchemy async session?

async with session.begin(): transaction scope - see async databases page.

Q: Testing async iterators?

Drive __anext__ until StopAsyncIteration or collect via async for in pytest-asyncio.

Q: async generator cleanup?

Use try/finally inside generator; consumers should async with aclosing(gen()) (3.10+).

Q: websocket loop?

async for message in ws: until connection closed - handle ConnectionClosed gracefully.

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
httpx streaming 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:

@asynccontextmanager mirrors sync contextmanager with await
async with calls __aenter__/__aexit__ awaitables
__aiter__/__anext__ power async for
StopAsyncIteration ends 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 with on async manager - does not await teardown. Fix: always async 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 for only.
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.

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+.

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
httpx streaming 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:

@asynccontextmanager mirrors sync contextmanager with await
async with calls __aenter__/__aexit__ awaitables
__aiter__/__anext__ power async for
StopAsyncIteration ends 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 with on async manager - does not await teardown. Fix: always async 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 for only.
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.

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+.

Async Context Managers & Iterators

Search Documentation

Async Context Managers & Iterators