All 15 Async Generators & Async Context Managers questions
More in Async Generators & Async Context Managers
How do you write an async context manager more concisely using @contextlib.asynccontextmanager? How does an async generator interact with try/finally for guaranteed cleanup, similar to regular generators? 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 do you handle backpressure when an async generator produces items faster than the consumer can process them? How would you implement timeout handling for an async generator's individual yields, so a stalled source doesn't hang the consumer forever? 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?
Advertisement
Python › Async Generators & Async Context Managers

How would you implement a custom async iterator class from scratch, without using an async generator function?

Advanced

Answer

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