Python Async Iterators and Async Generators

Iteration is Python's most natural way to process data. You use for loops to walk through lists, read files line by line, and consume generator pipelines that transform data without loading everything into memory at once. Async iterators and async generators bring the same model into async Python.

An async iterator replaces iter and next with __aiter__ and __anext__, and an async for loop replaces for. An async generator replaces yield with yield inside an async def, allowing it to await I/O operations between each yielded value. If you have read about iterators and generators, you understand the synchronous iteration model.

This article covers its async counterpart.

The need for async iteration arises naturally in real programs. When you stream rows from an async database driver, paginate through a REST API, or consume messages from a queue, the next value requires an async operation. A synchronous generator cannot await that operation.

An async generator can, and it yields each value as it arrives without blocking the event loop.

Async iterators: the protocol

An async iterable implements aiter(), which returns an async iterator. An async iterator implements anext(), which returns an awaitable that resolves to the next value or raises StopAsyncIteration when done.

python

import asyncio
 
class AsyncRange:
    def __init__(self, start, stop):
        self.start = start
        self.stop = stop
 
    def __aiter__(self):
        self.current = self.start
        return self
 
    async def __anext__(self):
        if self.current >= self.stop:
            raise StopAsyncIteration
        await asyncio.sleep(0.01)
        value = self.current
        self.current += 1
        return value
 
async def main():
    async for n in AsyncRange(0, 5):
        print(n)
 
asyncio.run(main())

Each number prints only after its own asyncio.sleep(0.01) inside anext completes, so the loop produces the same five values a synchronous range(5) would, just with an await between each one:

text

0
1
2
3
4

The async for loop calls aiter() once to get the iterator, then repeatedly awaits anext() until StopAsyncIteration is raised. Each anext() call is awaited, so the event loop can run other tasks between iterations.

Most real async iterators should also implement aiter() to return self, so they are both iterable and iterator. This matches the synchronous convention and makes them usable everywhere async iterables are expected.

Async generators: the simpler way

Writing an async iterator class is verbose. An async generator replaces the class with an async def function that uses yield:

python

import asyncio
 
async def async_range(start, stop):
    for i in range(start, stop):
        await asyncio.sleep(0.01)
        yield i
 
async def main():
    async for n in async_range(0, 5):
        print(n)
 
asyncio.run(main())

The async def with yield makes this an async generator function. Calling it returns an async generator object, which implements the async iterator protocol automatically. The await inside the generator pauses between yields, letting the event loop run other tasks.

Async generators are the right choice for almost all async iteration needs. The class-based protocol is only necessary when you need custom state management that does not fit cleanly into a generator function.

Streaming data from an async source

The canonical use case for async generators is streaming data from an async source, like paginated API responses:

python

import asyncio
 
async def paginated_api(base_url, pages):
    """Simulate fetching paginated results from an API."""
    for page in range(1, pages + 1):
        await asyncio.sleep(0.1)
        yield f"{base_url}?page={page}", [f"item-{page}-{i}" for i in range(3)]
 
async def list_all_items(base_url, pages):
    items = []
    async for url, page_items in paginated_api(base_url, pages):
        print(f"Fetched {url}")
        items.extend(page_items)
    return items
 
async def main():
    results = await list_all_items("https://api.example.com/data", 3)
    print(f"Total items: {len(results)}")
 
asyncio.run(main())

Each page is fetched asynchronously and yielded as it arrives. The consumer processes each page without waiting for all pages to be fetched. If each page takes 100 ms and there are 3 pages, the consumer starts processing the first page after 100 ms rather than 300 ms.

Async generator pipelines

Like synchronous generators, async generators compose into pipelines. One async generator transforms the output of another:

python

import asyncio
 
async def read_sensor_data():
    """Simulate reading from a sensor every 100 ms."""
    for reading in range(1, 6):
        await asyncio.sleep(0.1)
        yield reading
 
async def filter_significant(readings, threshold):
    """Pass through only readings above the threshold."""
    async for value in readings:
        if value >= threshold:
            yield value
 
async def format_output(readings):
    """Format readings for display."""
    async for value in readings:
        yield f"Sensor: {value}"

Each of these three functions only knows about the async iterable it receives as an argument, which is what lets main() chain them together into a pipeline without any of the stages knowing about the others:

python

async def main():
    raw = read_sensor_data()
    filtered = filter_significant(raw, threshold=3)
    formatted = format_output(filtered)
 
    async for message in formatted:
        print(message)
 
asyncio.run(main())

Readings 1 and 2 never reach the final print because filter_significant drops anything below the threshold of 3, so only the three qualifying sensor values flow all the way through to format_output:

text

Sensor: 3
Sensor: 4
Sensor: 5

Each stage in the pipeline is an independent async generator. Data flows through the pipeline one value at a time: read_sensor_data yields a value, filter_significant checks it and maybe yields it, format_output formats and yields it. At no point is the entire dataset materialized in memory.

This pattern is especially valuable for processing large streams from message queues, event logs, or real-time data sources where loading everything into a list would exhaust memory.

Async comprehensions

Python 3.6 introduced async comprehensions, which let you use async for inside list, set, and dict comprehensions, as well as generator expressions. They can only appear inside async def functions.

python

import asyncio
 
async def generate_values():
    for i in range(5):
        await asyncio.sleep(0.01)
        yield i * 10
 
async def main():
    # Async list comprehension
    values = [v async for v in generate_values()]
    print(values)
 
    # Async list comprehension with filtering
    filtered = [v async for v in generate_values() if v > 20]
    print(filtered)
 
asyncio.run(main())

The same async for syntax also works directly inside parentheses instead of square brackets, which creates an async generator expression without needing to define a separate named function first:

python

async def main():
    squares = (x * x async for x in generate_values())
    async for square in squares:
        print(square)

Async comprehensions are concise but collect all results into memory. For large streams, prefer async generator pipelines that process values one at a time.

Await inside comprehensions

You can also use await inside comprehensions (Python 3.6+), which is different from async for. An await in a comprehension awaits each value and collects the resolved results:

python

import asyncio
 
async def fetch(id):
    await asyncio.sleep(0.01)
    return f"Result {id}"
 
async def main():
    # Await multiple coroutines, collecting results
    results = [await fetch(i) for i in range(5)]
    print(results)
 
asyncio.run(main())

This runs fetch(0), awaits it, then runs fetch(1), awaits it, and so on. It is sequential, not concurrent. To run them concurrently, use gather():

python

async def main():
    results = await asyncio.gather(*[fetch(i) for i in range(5)])
    print(results)

Custom async iterables with cleanup

When an async iterator manages resources like network connections or file handles, it should implement aiter and anext on a class and provide cleanup when iteration is abandoned.

python

import asyncio
 
class AsyncFileReader:
    def __init__(self, filename):
        self.filename = filename
    async def __aenter__(self):
        await asyncio.sleep(0.01)
        self.lines = ["line 1\n", "line 2\n", "line 3\n"]
        self.index = 0
        return self
    async def __aexit__(self, *args):
        self.lines = None
        await asyncio.sleep(0.01)
    def __aiter__(self):
        return self
    async def __anext__(self):
        if self.index >= len(self.lines):
            raise StopAsyncIteration
        line = self.lines[self.index]
        self.index += 1
        return line

Combining __aenter__/__aexit__ with aiter/anext on the same class means a single async with block both sets up the resource and lets you iterate over it, with cleanup guaranteed even if the loop exits early:

python

async def main():
    async with AsyncFileReader("data.txt") as reader:
        async for line in reader:
            print(line, end="")
 
asyncio.run(main())

The async with statement ensures aexit is called when the block exits, just as with calls exit for synchronous context managers. This is covered in detail in the article on async context managers.

Practical use cases

Database row streaming. Async database drivers like asyncpg return async iterables for query results:

python

async def stream_users(conn):
    async for row in conn.cursor("SELECT * FROM users"):
        yield row

Message queue consumption. Async message brokers like Redis streams or Kafka return messages as async iterables, so a consumer can async for over incoming messages the same way it would over database rows:

python

async def consume_messages(client, topic):
    async for message in client.subscribe(topic):
        yield message

Paginated API consumption. REST APIs with cursor-based pagination map naturally to async generators, since each page fetch is itself an async operation and the generator can keep requesting the next page until the API stops returning one:

python

async def paginated_fetch(client, url):
    while url is not None:
        response = await client.get(url)
        data = response.json()
        for item in data["items"]:
            yield item
        url = data.get("next")

In each case, the async generator encapsulates the I/O pattern and exposes a clean async for interface to the consumer. The consumer processes items as they arrive without knowing whether they come from a database, a queue, or an API.