python async iterator: Practical Usage and Code Examples

When you need to iterate over data that arrives asynchronously—such as messages from a WebSocket, rows streaming from a database cursor, or chunks read from a network socket—a regular iterator blocks the event loop while waiting for each item. Python's async iterator protocol solves this by allowing async for to await each next value without freezing other tasks. This article explains how to build and use a python async iterator correctly, covering the protocol, async generators, error handling, and resource cleanup.

The Async Iterable Protocol

An async iterator is any object that implements __aiter__() and __anext__(). The __aiter__() method must return an async iterator object, and __anext__() must return an awaitable that resolves to the next value or raises StopAsyncIteration when exhausted. The async for loop calls __aiter__() once, then repeatedly awaits __anext__() until the exception is raised.

Here is a minimal manual implementation that yields numbers from a list with a small artificial delay:

import asyncio

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.1)
        self.current += 1
        return self.current

Using this class with async for is straightforward:

async def main():
    async for number in AsyncCounter(3):
        print(number)

asyncio.run(main())

This works, but writing a full class for every asynchronous sequence is verbose. In practice, you rarely need to implement the protocol manually because Python provides a much more concise way: async generators.

Creating Async Iterators with Async Generators

An async generator is a function defined with async def that contains at least one yield statement. When called, it returns an async generator object that automatically implements __aiter__() and __anext__(). The same counter becomes a few lines:

async def async_counter(limit):
    for i in range(limit):
        await asyncio.sleep(0.1)
        yield i + 1

You can consume it with async for exactly as before. The key difference from a synchronous generator is that the generator's execution is suspended at each yield and can be resumed only by awaiting __anext__(). This allows you to perform asynchronous operations between yields, such as reading from a socket or waiting for a network response.

Because the async generator object is itself an async iterator, it can be passed directly to functions that expect one. For example, you can wrap it with asyncio.as_completed or use it in a list comprehension with async for inside an async function.

Using async for Loops Effectively

The async for loop is the natural consumer of an async iterator. It works only inside an async def function. The loop awaits each __anext__() call, so the event loop can run other tasks while waiting for data. This is the main advantage over a regular for loop, which would block the entire thread.

Consider a scenario where you fetch data from multiple sources concurrently. You can start several async iterators and interleave their consumption using asyncio.gather or by iterating over them in separate tasks. However, async for itself processes items sequentially; if you need concurrent processing, you must create separate tasks for each iterator.

A common pattern is to use an async generator to wrap a paginated API. Each yield fetches the next page asynchronously, and the consumer processes items as they arrive:

async def fetch_pages():
    page = 1
    while True:
        data = await api_get(f"items?page={page}")
        if not data:
            break
        yield data
        page += 1

The consumer can then process each page without blocking other tasks. This pattern is especially useful for data pipelines where you want to avoid loading the entire dataset into memory.

Handling Errors and Cancellation

Async iterators can raise exceptions during iteration. If an exception is raised inside the async generator, it propagates to the async for loop, and the generator is automatically closed. You can catch the exception in the consumer to handle it gracefully:

try:
    async for item in async_generator():
        process(item)
except SomeError:
    handle_error()

When an async generator is closed prematurely—for example, because the consumer breaks out of the loop or the task is cancelled—Python calls the generator's aclose() method. This is where you should release resources like network connections or file handles. You can implement cleanup by wrapping the generator body in a try/finally block:

async def async_generator():
    conn = await open_connection()
    try:
        while True:
            item = await conn.receive()
            yield item
    finally:
        await conn.close()

If you are implementing the protocol manually, you must also implement asend() and athrow() if you need to send values or throw exceptions into the iterator. For most use cases, async generators handle these automatically, but understanding aclose() is important for resource management.

Performance and Resource Management

Async iterators are lazy by nature: they produce items only when requested. This means you can process data streams that would not fit in memory, such as reading a large file line by line or consuming a high-volume message queue. The tradeoff is that each __anext__() call involves an await, which adds a small overhead compared to a synchronous iterator. In practice, the overhead is negligible when the iteration involves I/O, which is the typical use case.

One performance pitfall is creating a new async iterator for every item in a loop. For example, if you call an async generator function inside a loop, each call creates a new generator object, which may open new connections or allocate new resources. Instead, create the iterator once and reuse it:

# Bad: creates a new generator each iteration
for _ in range(10):
    async for item in fetch_items():
        process(item)

# Good: reuse the same generator
async for item in fetch_items():
    process(item)

Another consideration is memory usage when buffering. If you collect items into a list, you lose the memory advantage. Use streaming processing whenever possible, and rely on the async iterator's lazy evaluation to keep memory bounded.

Common Mistakes and Edge Cases

A frequent mistake is trying to use async for outside an async function. The async for loop is a syntax error in synchronous code. Similarly, calling next() on an async iterator fails because next() expects a synchronous iterator. You must use await anext() (Python 3.10+) or await iterator.__anext__() manually.

Another edge case is mixing synchronous and asynchronous iteration. If you have an async generator that yields values that themselves are awaitables, you must explicitly await them inside the loop:

async for awaitable in async_gen():
    value = await awaitable
    process(value)

This is easy to forget when the generator yields coroutines from other async functions.

When implementing the protocol manually, remember that __anext__ must return an awaitable. A common mistake is returning a value directly instead of a coroutine. Also, __aiter__ can return self, but it can also return a different async iterator object if you want to support multiple iterations over the same data. However, most async iterators are single-use, like async generators.

Finally, be aware that async for does not automatically handle backpressure. If the producer generates items faster than the consumer processes them, the event loop will queue the coroutines, potentially increasing memory usage. For high-throughput pipelines, consider using a bounded queue or a flow-control mechanism to limit the number of pending __anext__ calls.