All 15 Async Generators & Async Context Managers questions
More in Async Generators & Async Context Managers
How do you build an async generator pipeline, chaining one async generator's output into another, like a synchronous generator pipeline? How do async comprehensions work, combining 'async for' directly inside a list/set/dict comprehension? How would you implement a custom async iterator class from scratch, without using an async generator function? How do you handle backpressure when an async generator produces items faster than the consumer can process them? How do async generators interact with asyncio.Queue to implement a producer-consumer pattern across separate tasks? How would you implement an async context manager that also supports being used as a decorator, similar to contextlib.contextmanager's decorator support? How do you properly propagate cancellation into an async generator when the consuming task is cancelled mid-iteration? How would you implement a 'fan-out, fan-in' pattern using multiple async generators consumed concurrently and merged into one stream? How do you correctly test async generators and async context managers using pytest and pytest-asyncio?
Advertisement
Python › Async Generators & Async Context Managers

How would you implement timeout handling for an async generator's individual yields, so a stalled source doesn't hang the consumer forever?

Advanced

Answer

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
All 15 Async Generators & Async Context Managers questions