Async Generators & Async Context Managers
15 questions found
What is an async generator, and how does it differ from a regular generator?
Beginner
An async generator is a function defined with 'async def' that uses 'yield' inside it, producing an object supporting the asynchronous iteration protocol (__anext__) instead of the regular one — it lets each item be produced after an 'await', unlike a regular generator which is fully synchronous.
async def fetch_pages():
for i in range(3):
await asyncio.sleep(0.1) # simulate async work
yield f"page-{i}"
Real-world example
Streaming paginated API results one page at a time, awaiting each network call before yielding the next page.
Common follow-ups: How do you actually consume an async generator's yielded values?
Concurrency (asyncio/threading/multiprocessing)
How do you consume an async generator using 'async for'?
Beginner
'async for' is the asynchronous equivalent of a regular 'for' loop — it awaits each call to the generator's __anext__() method, pausing until the next value is ready, and must itself be used inside an 'async def' function.
async def main():
async for page in fetch_pages():
print(page)
asyncio.run(main())
Real-world example
Processing streamed results from an async API client one item at a time as they arrive.
Common follow-ups: Can 'async for' be used outside of an async function?
Concurrency (asyncio/threading/multiprocessing)
What is an async context manager, and which two dunder methods does it require?
Beginner
An async context manager implements __aenter__ and __aexit__ (both coroutines, defined with 'async def') instead of the synchronous __enter__/__exit__ — used with 'async with' to manage a resource whose setup or teardown itself requires awaiting, like an async database connection.
class AsyncConnection:
async def __aenter__(self):
await self.connect()
return self
async def __aexit__(self, exc_type, exc, tb):
await self.disconnect()
async def main():
async with AsyncConnection() as conn:
await conn.query("SELECT 1")
Real-world example
Managing an async database or HTTP client session where opening/closing the connection are themselves async operations.
Common follow-ups: What happens to __aexit__ if an exception is raised inside the 'async with' block?
Concurrency (asyncio/threading/multiprocessing)
How do you write an async context manager more concisely using @contextlib.asynccontextmanager?
Intermediate
asynccontextmanager decorates a single async generator function: code before the 'yield' runs as __aenter__, the yielded value becomes the 'as' target, and code after 'yield' (typically in a try/finally) runs as __aexit__ — avoiding the need to write a full class with both dunder methods.
from contextlib import asynccontextmanager
@asynccontextmanager
async def async_connection():
conn = await connect()
try:
yield conn
finally:
await conn.close()
async def main():
async with async_connection() as conn:
await conn.query("SELECT 1")
Real-world example
Writing a lightweight, function-based async resource manager without the boilerplate of a full __aenter__/__aexit__ class.
Common follow-ups: How does this asynccontextmanager decorator compare to the synchronous @contextlib.contextmanager?
functools & Functional Programming Tools
How does an async generator interact with try/finally for guaranteed cleanup, similar to regular generators?
Intermediate
Just like a regular generator, wrapping the yield in a try/finally inside an async generator guarantees the finally block runs when the generator is exhausted, explicitly closed via aclose(), or garbage collected — ensuring resources like an open connection are properly released even if the consumer stops iterating early.
async def read_stream(conn):
try:
while True:
chunk = await conn.read_chunk()
if not chunk:
break
yield chunk
finally:
await conn.close() # runs even if consumer breaks out of 'async for' early
Real-world example
Ensuring a network connection or file handle used inside an async generator is properly closed even if the consumer stops early.
Common follow-ups: What method would you call to explicitly close an async generator without fully iterating it?
Exception Handling
How do you build an async generator pipeline, chaining one async generator's output into another, like a synchronous generator pipeline?
Intermediate
Define each stage as its own async generator that takes an async iterable as input and 'async for's over it, yielding transformed values — chaining multiple stages together creates a fully lazy, asynchronous processing pipeline, mirroring how synchronous generators compose.
async def double(source):
async for item in source:
yield item * 2
async def only_even(source):
async for item in source:
if item % 2 == 0:
yield item
async def main():
async for value in only_even(double(numbers_stream())):
print(value)
Real-world example
Building a lazy, asynchronous ETL-style pipeline that transforms and filters streamed data without buffering it all in memory.
Common follow-ups: How does this pipeline pattern avoid loading the entire dataset into memory at once?
Comprehensions & Generators
How do async comprehensions work, combining 'async for' directly inside a list/set/dict comprehension?
Intermediate
Python allows 'async for' (and 'if') clauses directly inside a comprehension, as long as it's evaluated inside an async function — this asynchronously iterates the source, awaiting each step, while still producing a regular (synchronous) list/set/dict as the final result.
async def main():
results = [page async for page in fetch_pages()]
print(results) # a regular list, built by asynchronously iterating fetch_pages()
Real-world example
Collecting all results from an async generator into a plain list when you know the full result set is manageable in memory.
Common follow-ups: Can you combine 'await' expressions ALSO inside the same async comprehension, alongside 'async for'?
Comprehensions & Generators
How would you implement a custom async iterator class from scratch, without using an async generator function?
Advanced
Implement __aiter__ (returning self) and __anext__ (an async method that returns the next value or raises StopAsyncIteration when exhausted) — this is the lower-level protocol that async generator functions are automatically compiled down to by the interpreter.
class AsyncCounter:
def __init__(self, limit):
self.limit = limit
self.current = 0
def __aiter__(self):
return self
async def __anext__(self):
if self.current >= self.limit:
raise StopAsyncIteration
await asyncio.sleep(0.01)
self.current += 1
return self.current
Real-world example
Building a custom async iterable class with more control than an async generator function provides, like exposing extra methods or state.
Common follow-ups: Why would you choose to write a full __aiter__/__anext__ class instead of a simpler async generator function?
Iterators & the Iterator Protocol
How do you handle backpressure when an async generator produces items faster than the consumer can process them?
Advanced
Since 'async for' inherently pauses the PRODUCER (the generator's execution) at each yield until the consumer calls __anext__() again, async generators are naturally backpressure-aware — the producer literally cannot get ahead of the consumer, unlike a callback-based push model where you'd need an explicit buffer or queue.
async def slow_consumer(source):
async for item in source:
await asyncio.sleep(1) # slow processing
process(item)
# the producer naturally waits here until we call __anext__ again
Real-world example
Processing a fast-arriving async data stream (like a websocket feed) without needing to manually implement a bounded buffer or queue.
Common follow-ups: How would you add EXPLICIT buffering on top of this if you wanted the producer to run ahead by a controlled amount?
Concurrency (asyncio/threading/multiprocessing)
How would you implement timeout handling for an async generator's individual yields, so a stalled source doesn't hang the consumer forever?
Advanced
Wrap each call to the generator's __anext__() in asyncio.wait_for() with a timeout, catching asyncio.TimeoutError if a specific item takes too long to arrive — since 'async for' itself doesn't provide a built-in per-item timeout mechanism.
async def with_timeout(source, timeout):
it = source.__aiter__()
while True:
try:
yield await asyncio.wait_for(it.__anext__(), timeout)
except StopAsyncIteration:
break
async def main():
async for item in with_timeout(slow_stream(), timeout=2.0):
print(item)
Real-world example
Cutting off a stalled network stream or SSE connection after a set idle period, instead of hanging indefinitely.
Common follow-ups: What exception does asyncio.wait_for() raise when the timeout expires, and how do you distinguish it from StopAsyncIteration?
Exception Handling