Async Context Managers & Iterators
async with and async for extend the context-manager and iterator protocols with asynchronous __aenter__/__aexit__ and __aiter__/__anext__ methods, for resources and sequences whose setup, teardown, or next-item retrieval must itself be awaited.
Regular context managers (with) and iterators (for) work with synchronous protocol methods (__enter__/__exit__, __iter__/__next__) that must complete instantly, without waiting on I/O. When acquiring a resource, releasing it, or producing the next item genuinely requires an asynchronous operation -- opening a network connection, flushing a socket, awaiting the next message from a queue -- Python's async with and async for statements, paired with the __aenter__/__aexit__ and __aiter__/__anext__ dunder methods, provide the async equivalents.
async with
async with expr as name: is the asynchronous counterpart to with. It requires expr to be an async context manager -- an object implementing __aenter__ and __aexit__ as coroutine methods (defined with async def) rather than ordinary methods. Just like with, it guarantees __aexit__ runs on the way out of the block, including when an exception propagates through it -- but both the entry and exit steps are awaited, so they can themselves perform I/O.
__aenter__ and __aexit__ are coroutines here, so opening and closing the connection can genuinely await I/O rather than blocking synchronously.
import asyncio
class AsyncConnection:
async def __aenter__(self) -> "AsyncConnection":
print("opening connection...")
await asyncio.sleep(0.1) # simulate an async handshake
print("connection open")
return self
async def __aexit__(self, exc_type, exc_value, traceback) -> None:
print("closing connection...")
await asyncio.sleep(0.1) # simulate an async close
print("connection closed")
async def query(self, sql: str) -> str:
await asyncio.sleep(0.05)
return f"result of: {sql}"
async def main() -> None:
async with AsyncConnection() as conn:
result = await conn.query("SELECT 1")
print(result)
asyncio.run(main())__aenter__ and __aexit__
__aenter__(self) is called (and awaited) when entering the async with block; its return value becomes the as name target. __aexit__(self, exc_type, exc_value, traceback) is called (and awaited) when leaving the block, whether normally or due to an exception, with the same signature convention as synchronous __exit__: returning a truthy value suppresses a propagating exception, and returning None/falsy lets it propagate.
The contextlib standard-library module provides @asynccontextmanager, an async equivalent of @contextmanager, letting you write an async context manager as a single async generator function with one yield instead of a full class with __aenter__/__aexit__ -- convenient for simple cases.
from contextlib import asynccontextmanager
import asyncio
@asynccontextmanager
async def async_resource(name: str):
print(f"acquiring {name}")
await asyncio.sleep(0.1)
try:
yield f"handle-to-{name}"
finally:
print(f"releasing {name}")
await asyncio.sleep(0.1)
async def main() -> None:
async with async_resource("db") as handle:
print(f"using {handle}")
asyncio.run(main())async for
async for item in expr: is the asynchronous counterpart to for. It requires expr to be an async iterable -- an object implementing __aiter__, which returns an async iterator implementing __anext__ as a coroutine method. On each iteration, async for awaits __anext__(); the loop ends when that call raises StopAsyncIteration (the async equivalent of StopIteration).
message_stream is an async generator (async def + yield); it automatically implements the async iterator protocol, so it can be driven directly by async for.
import asyncio
async def message_stream(count: int):
for i in range(count):
await asyncio.sleep(0.2) # simulate waiting for the next message
yield f"message {i}"
async def main() -> None:
async for msg in message_stream(3):
print(msg)
asyncio.run(main())
# message 0 (after ~0.2s)
# message 1 (after ~0.4s)
# message 2 (after ~0.6s)__aiter__ and __anext__
__aiter__(self) returns the async iterator object itself (or a separate one) and is typically defined as an ordinary (non-async) method returning self, mirroring how __iter__ usually works. __anext__(self) is a coroutine method that produces the next value on each call, and raises StopAsyncIteration when the sequence is exhausted -- implemented by hand, this looks like a manual, class-based version of what an async generator function gives you automatically.
Most of the time, an async generator function (async def containing yield) is simpler and less error-prone than hand-writing __aiter__/__anext__ on a class -- reach for the manual protocol mainly when you need extra state or behavior an async generator function can't easily express (e.g. also supporting async with on the same object).
import asyncio
class Countdown:
def __init__(self, start: int) -> None:
self.current = start
def __aiter__(self) -> "Countdown":
return self
async def __anext__(self) -> int:
if self.current <= 0:
raise StopAsyncIteration
await asyncio.sleep(0.1) # simulate an async delay per item
value = self.current
self.current -= 1
return value
async def main() -> None:
async for n in Countdown(3):
print(n) # 3, 2, 1 -- each after ~0.1s
asyncio.run(main())Async Generators
An async def function containing yield is an async generator function; calling it returns an async generator object, which automatically satisfies the async iterator protocol (it has __aiter__ and __anext__ built in). Inside an async generator, you can freely mix yield with await, letting you produce a value only after some asynchronous wait -- exactly the message_stream pattern shown above. Async generators are distinct from regular generators (yield without async def) and from coroutines (async def without yield, producing a single result) -- know which of the three you're looking at from the async/yield combination present.
Practical Usage
async with shows up wherever a resource's setup/teardown involves I/O: async database connection pools, async HTTP client sessions (as provided by third-party libraries like aiohttp or httpx), and asyncio's own synchronization primitives (asyncio.Lock, Semaphore all support async with for acquire/release). async for shows up wherever a sequence's items arrive over time via I/O: reading lines from an async stream, paginating results from an API one page at a time, or consuming messages from an async queue.
import asyncio
async def paginated_results(pages: int):
for page in range(pages):
await asyncio.sleep(0.1) # simulate an API call for this page
yield [f"item-{page}-{i}" for i in range(3)]
async def main() -> None:
async for page_items in paginated_results(2):
for item in page_items:
print(item)
asyncio.run(main())Edge Cases and Pitfalls
async with/async for require their target to implement the async protocol methods -- using them on an ordinary synchronous context manager or iterable raises a TypeError (e.g. 'object does not support the asynchronous context manager protocol'). Conversely, __aenter__/__anext__ etc. cannot be driven by plain with/for, since those expect ordinary (non-awaitable) return values, not coroutines. Async generators must be closed properly when abandoned early (e.g. breaking out of an async for loop) -- Python handles this automatically via garbage collection in most cases, but explicit .aclose() (or async with contextlib.aclosing(gen):) is safer when deterministic cleanup timing matters.
Complete Example
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def async_session(name: str):
print(f"[{name}] session opened")
try:
yield name
finally:
print(f"[{name}] session closed")
async def event_feed(count: int):
for i in range(count):
await asyncio.sleep(0.1)
yield {"event_id": i, "type": "ping"}
async def main() -> None:
async with async_session("feed-consumer") as session:
events = []
async for event in event_feed(3):
print(f"[{session}] received {event}")
events.append(event)
print(f"[{session}] total events: {len(events)}")
asyncio.run(main())Mental Model
Think of async with as checking into and out of a hotel where both check-in and check-out involve a genuine wait (verifying ID, settling a bill) rather than an instant handoff -- you can't skip ahead until each step actually finishes, but the front desk can help other guests while you wait. Think of async for as a delivery subscription: each new item arrives only after some wait (a courier en route), rather than an entire shelf of items being available to grab all at once the way a plain list is.
Quick Reference
async with obj as name: ... - requires __aenter__/__aexit__ (both coroutines)
async for item in obj: ... - requires __aiter__ + __anext__ (coroutine)
StopAsyncIteration - raised by __anext__ to end an async for loop
async def f(): yield x - async generator: auto-implements async iterator protocol
@asynccontextmanager - contextlib helper: write async CM as one-yield async generator
asyncio.Lock/Semaphore - already support async with directlyKey Takeaways
async with requires an async context manager (__aenter__/__aexit__ as coroutines) for resources whose setup/teardown itself needs to be awaited.
async for requires an async iterable (__aiter__ returning an object with a coroutine __anext__) for sequences whose next item requires awaiting.
__anext__ raises StopAsyncIteration to end the loop, mirroring StopIteration for regular iterators.
An async def function containing yield is an async generator and automatically satisfies the async iterator protocol -- usually simpler than hand-writing a class.
contextlib.asynccontextmanager lets you write a simple async context manager as a single-yield async generator function instead of a full class.
Sync and async protocols don't mix: a plain with/for cannot drive __aenter__/__anext__, and async with/async for cannot drive plain __enter__/__next__.
class Resource: def __enter__(self): ... # plain, synchronous protocol def __exit__(self, *a): ... async def main(): async with Resource() as r: # TypeError: not an async context manager ...
class Resource: async def __aenter__(self): ... async def __aexit__(self, *a): ... async def main(): async with Resource() as r: ...
async with requires __aenter__/__aexit__ defined as coroutine methods (async def); a class implementing only the synchronous __enter__/__exit__ protocol does not satisfy it.
class Items: def __iter__(self): return self def __next__(self): ... # plain, synchronous protocol async def main(): async for item in Items(): # TypeError: not an async iterable ...
class Items: def __aiter__(self): return self async def __anext__(self): ... async def main(): async for item in Items(): ...
async for requires __aiter__ + a coroutine __anext__; an object only implementing the synchronous __iter__/__next__ protocol cannot be driven by async for.
class Counter: def __aiter__(self): return self async def __anext__(self): if self.n <= 0: raise StopIteration # wrong exception for async iteration ...
class Counter: def __aiter__(self): return self async def __anext__(self): if self.n <= 0: raise StopAsyncIteration # correct signal to end an async for loop ...
__anext__ must raise StopAsyncIteration to end iteration; raising the synchronous StopIteration inside a coroutine is treated as an error, not a normal end-of-sequence signal.
What must an object implement to be usable with async with, and how do these differ from the synchronous context manager protocol?
It must implement __aenter__ and __aexit__ as coroutine methods (defined with async def); unlike the synchronous __enter__/__exit__, both are awaited when entering and leaving the block, so they can perform asynchronous operations like I/O.
What must an object implement to be usable with async for?
__aiter__, which returns an async iterator, and __anext__ on that iterator, defined as a coroutine method; async for awaits __anext__() on each iteration until it raises StopAsyncIteration.
What exception signals the end of an async for loop, and how does it differ from the synchronous case?
StopAsyncIteration, raised inside __anext__; it's the async-specific counterpart to StopIteration, which is what plain __next__ raises to end a regular for loop.
What is an async generator, and how do you create one?
An object automatically satisfying the async iterator protocol, created by defining an async def function that contains at least one yield statement; calling it returns the async generator object, drivable with async for.
What does contextlib.asynccontextmanager let you avoid writing?
A full class with separate __aenter__ and __aexit__ coroutine methods -- instead you write a single async generator function with one yield, where code before the yield is the setup and code after (typically in a finally) is the teardown.
Why can't a plain with statement be used on an object that only implements __aenter__/__aexit__?
with expects __enter__/__exit__ to return ordinary values immediately; __aenter__/__aexit__ are coroutines that must be awaited, which plain with does not know how to do -- only async with drives coroutine-based protocol methods.
Why might you need to explicitly close an async generator that's abandoned mid-iteration (e.g. via break)?
To ensure any cleanup code in the generator (e.g. a finally block) runs deterministically rather than whenever garbage collection happens to reclaim it; calling .aclose() (or using contextlib.aclosing) triggers that cleanup promptly.
Write an async context manager class that times how long its block took, using __aenter__/__aexit__.
import asyncio import time class AsyncTimer: async def __aenter__(self) -> "AsyncTimer": self._start = time.perf_counter() return self async def __aexit__(self, exc_type, exc_value, traceback) -> None: elapsed = time.perf_counter() - self._start print(f"elapsed: {elapsed:.2f}s") async def main() -> None: async with AsyncTimer(): await asyncio.sleep(0.3) asyncio.run(main()) # prints "elapsed: 0.30s"
__aenter__ records the start time when the block begins; __aexit__ computes and prints the elapsed time when the block ends, whether it exits normally or via an exception. Both are coroutine methods, so they're compatible with async with.
What happens when this code runs?
It raises a RuntimeError (StopIteration bubbling out of a coroutine's __anext__ is converted into a RuntimeError rather than treated as a normal end-of-iteration signal) instead of cleanly printing 1, 2, 3. __anext__ must raise StopAsyncIteration, not StopIteration, to properly end an async for loop.
Write an async generator that yields numbers with a delay between each, and consume it with async for inside an async context manager that logs entry/exit.
import asyncio from contextlib import asynccontextmanager @asynccontextmanager async def logged_block(name: str): print(f"[{name}] entering") try: yield finally: print(f"[{name}] exiting") async def delayed_numbers(n: int): for i in range(1, n + 1): await asyncio.sleep(0.1) yield i async def main() -> None: async with logged_block("numbers"): async for num in delayed_numbers(3): print(num) asyncio.run(main()) # [numbers] entering # 1 # 2 # 3 # [numbers] exiting logged_block, built with @asynccontextmanager, prints on entry and (via finally) on exit around the whole block; delayed_numbers is an async generator that awaits a short delay before each yield, and async for drives it one value at a time inside the logged block.
class Numbers:
def __init__(self, limit):
self.limit = limit
self.n = 0
def __aiter__(self):
return self
async def __anext__(self):
if self.n >= self.limit:
raise StopIteration
self.n += 1
return self.n
async def main():
async for x in Numbers(3):
print(x)
import asyncio
asyncio.run(main())