Asynchronous Iterators and Iterables in Python

Async Iterators

Python’s documentation defines asynchronous iterators, or async iterators for short, as the following:

An object that implements the .__aiter__() and .__anext__() [special] methods. .__anext__() must return an awaitable object. [An] async for [loop] resolves the awaitables returned by an asynchronous iterator’s .__anext__() method until it raises a StopAsyncIteration exception. (Source)

Similar to regular iterators that must implement .__iter__() and .__next__(), async iterators must implement .__aiter__() and .__anext__(). In regular iterators, the .__iter__() method usually returns the iterator itself. This is also true for async iterators.

To continue with this parallelism, in regular iterators, the .__next__() method must return the next object for the iteration. In async iterators, the .__anext__() method must return the next object, which must be awaitable.

Python defines awaitable objects as described in the quote below:

An object that can be used in an await expression. [It] can be a coroutine or an object with an .__await__() method. (Source)

In practice, a quick way to make an awaitable object in Python is to call an asynchronous function. You define this type of function with the async def keyword construct. This call creates a coroutine object.

When the data stream runs out of data, the method must raise a StopAsyncIteration exception to end the asynchronous iteration process.

Here’s an example of an async iterator that allows iterating over a range of numbers asynchronously:

import asyncio

class AsyncRange:
    def __init__(self, start, end):
        self.start = start
        self.end = end

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.start < self.end:
            await asyncio.sleep(0.5)
            value = self.start
            self.start += 1
            return value
        else:
            raise StopAsyncIteration

async def main():
    async for i in AsyncRange(0, 5):
        print(i)

asyncio.run(main())

In the .__aiter__() method, you return self, which is the current object—the iterator itself. In the .__anext__() method, you generate a number from the range between .start and .end.

To simulate the required awaitable object, you use the asyncio.sleep() function with a delay of 0.5 seconds in an await statement. When the range is covered, you raise a StopAsyncIteration exception to finish the iteration.

When you run the script, you get the following output:

$ python async_range_v1.py
0
1
2
3
4

In this example, you’ll get each number after waiting half a second, which is congruent with the asynchronous iteration.

The above example is a quick first look at async iterators and how to define them. You’ll learn more about the .__aiter__() and .__anext__() methods and related concepts when you get to the sections on creating async iterators. Now, it’s time to learn the basics of async iterables.

Async Iterables

When it comes to async iterables, the Python documentation says the following:

An object, that can be used in an async for statement. Must return an asynchronous iterator from its .__aiter__() method. (Source)

In practice, an object only needs an .__aiter__() method that returns an async iterator to be iterable. It doesn’t need the .__anext__() method.

Here’s how you’d modify the AsyncRange to be an async iterable rather than an iterator:

import asyncio

class AsyncRange:
    def __init__(self, start, end):
        self.data = range(start, end)

    async def __aiter__(self):
        for i in self.data:
            await asyncio.sleep(0.5)
            yield i

async def main():
    async for i in AsyncRange(0, 10):
        print(i)

asyncio.run(main())

This new implementation of your AsyncRange class is more concise than the previous one. It just has the .__aiter__() method, which yields numbers on demand. Again, you use the asyncio.sleep() function to simulate the awaitable object.

Here’s how the class works:

$ python async_range_v2.py
0
1
2
3
4

You get the same result as in the previous section, but instead of using an async iterator, you use an iterable.

With this quick background on async iterators and iterables, you can now dive deeper into how asynchronous iteration works and why you’d want to use it in your code.

Async Iteration

In Python, asynchronous iteration refers to traversing asynchronous iterables using async for loops. Under the hood, async for loops rely on async iterators to control the iteration process. Asynchronous iteration allows you to perform non-blocking operations within the loop.

Async iteration enables you to handle I/O-bound tasks efficiently and makes it possible to run tasks concurrently. Common I/O-bound tasks include the following:

• File system operations, such as reading and writing files and accessing a file’s metadata like its size, creation date, and modification date.
• Network operations, such as HTTP requests and socket communication.
• Database operations, such as running SQL queries for CRUD (create, read, update, and delete) operations.
• User input and output operations, such as reading the user input from the keyboard, mouse, or other input device and displaying output to the screen. These operations are critical in GUI (graphical user interface) applications, where rendering the interface can be resource-intensive.
• External device communication, such as interacting with external sensors, printers, or other peripherals connected to serial or parallel ports.

In Python, asynchronous code runs in an event loop, which you typically start with the asyncio.run() function.

When you iterate over an async iterable using an async for loop, the loop gives control back to the event loop after each cycle so that other asynchronous tasks can run. This type of iteration is non-blocking because it doesn’t block the app’s execution while the loop is running.

Asynchronous iterators and iterables allow for asynchronous iteration, which lets you perform tasks concurrently.

Concurrency allows multiple tasks to progress by sharing time on the same CPU core or to run in parallel using multiple CPU cores. This programming technique can help you make your code more efficient. It also allows you to prevent blocking your program’s execution with time-consuming tasks like the ones listed above.

Asynchronous programming is a specific type of concurrency based on non-blocking operations and event-driven execution. That’s why async code runs in a main event loop, which takes care of handling asynchronous events.

In your asynchronous programming adventure in Python, you’ll probably be required to create your own asynchronous iterators and iterables. In practice, the preferred way to do this is using async generator iterators, which is the topic of the following section.

The .__aiter__() Method

When you create async iterators, the .__aiter__() method must be a regular method that immediately returns an async iterator object. The typical implementation of this method in an async iterator looks something like this:

class AsyncIterator:
    def __aiter__(self):
        return self

There isn’t much to this implementation. You define the method as a regular instance method and return self, which holds the current object, and the object is the iterator.

When creating async iterables, you only need to implement the .__aiter__() method for the iterable to work. However, in this case, the method will have a more elaborate implementation that returns a proper async iterator object that yields items on demand.

In practice, you’ll often code .__aiter__() as an async generator function with the yield statement. For example, say that you want to create an async iterable to process large files. In this situation, you can end up with the following code:

import asyncio

import aiofiles

class AsyncFileIterable:
    def __init__(self, path, chunk_size=1024):
        self.path = path
        self.chunk_size = chunk_size

    async def __aiter__(self):
        async with aiofiles.open(self.path, mode="rb") as file:
            while True:
                chunk = await file.read(self.chunk_size)
                if not chunk:
                    break
                yield chunk

async def main():
    async for chunk in AsyncFileIterable("large-file.md"):
        # Process the file chunk here...
        await asyncio.sleep(0.2)
        print(chunk.decode("utf-8"))

asyncio.run(main())

In this example, the AsyncFileIterable class implements only the .__aiter__() method as an async generator function. The method opens the input file and reads it in chunks. Then, it yields file chunks on demand. With this iterator, you can process large files in chunks without blocking the script’s execution. That’s what you simulate in the script’s main() function.

Go ahead and run the script against one of your large files. To do this, update the path to your large file when you instantiate AsyncFileIterable in the main() function.

Another way to write the .__aiter__() method is to make it return an existing async iterator:

class AsyncIterable:
    def __aiter__(self):
        return AsyncIterator(self)

class AsyncIterator:
    def __init__(self, iterable):
        self.iterable = iterable

    def __aiter__(self):
        return self

    async def __anext__(self):
        ...

In this example, the AsyncIterable class returns an instance of AsyncIterator from its .__aiter__() method.

The .__anext__() Method

Only async iterators need the .__anext__() method. This method should be async def because it needs to perform asynchronous operations to fetch the next item during iteration. So, the method should generally look something like this:

async def __anext__(self):
    ...

The .__anext__() method must return an awaitable object. It can be a coroutine object or an object with an .__await__() method.

Another characteristic of .__anext__() is that it has to raise a StopAsyncIteration exception when the data stream is exhausted or consumed. This exception will tell Python to terminate the iteration process.

Here’s an example of creating an async iterator to process a large file in chunks. It works the same as the example in the previous section, but instead of using an async iterable, it uses an async iterator that implements both the .__aiter__() and .__anext__() methods:

import asyncio

import aiofiles

class AsyncFileIterator:
    def __init__(self, path, chunk_size=1024):
        self.path = path
        self.chunk_size = chunk_size
        self.file = None

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.file is None:
            self.file = await aiofiles.open(self.path, mode="rb")
        chunk = await self.file.read(self.chunk_size)
        if not chunk:
            await self.file.close()
            raise StopAsyncIteration
        return chunk

async def main():
    async for chunk in AsyncFileIterator("large-file.md"):
        # Process the file chunk here...
        await asyncio.sleep(0.2)
        print(chunk.decode("utf-8"))

asyncio.run(main())

In this example, the .__aiter__() method provides the minimal required implementation of an async iterator. It just returns self, which is the current object—the iterator itself.

Then, you define the .__anext__() method. First, you open the file asynchronously. Note that you can’t use an async with statement here because, if you did, you’d be opening and closing the file in every call to .__anext__(), and your code wouldn’t work.

Next, you read a chunk of the target file, which is your awaitable object. The second conditional statement checks whether the chunk holds data. If not, then you close the file and raise the StopAsyncIteration exception to signal that the data is exhausted. Finally, you return the awaitable object, chunk.

In main(), you iterate over the file’s chunks and process them. Go ahead and run the script. You’ll get the same result as in the previous section.

The Built-in anext() Function

You can use the built-in anext() function to traverse an async iterator one item at a time in a controlled way. It’s particularly useful when you need more granular control over the iteration process. For example, you may need to skip a few items from the iterator before getting to the data that you want to process.

Consider the following code example where you create an async iterator to process CSV files:

import asyncio
import csv

import aiofiles

class AsyncCSVIterator:
    def __init__(self, path):
        self.path = path
        self.reader = None

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.reader is None:
            async with aiofiles.open(self.path, mode="r") as file:
                lines = await file.readlines()
                self.reader = csv.reader(lines)
        try:
            return next(self.reader)
        except StopIteration:
            raise StopAsyncIteration

async def main():
    csv_iter = AsyncCSVIterator("data.csv")
    # Skip the headers
    await anext(csv_iter)
    # Process the rest of the rows
    async for row in csv_iter:
        print(row)

asyncio.run(main())

In this example, the AsyncCSVIterator reads a CSV file’s content once in the .__anext__() method. The reading task runs asynchronously. Then, it returns a single line at a time.

In main(), you use anext() to skip the first row of the CSV file. This line typically contains the headers for your data. Then, you start a loop over the rest of the rows, which hold the actual data.

The anext() function can also help when you must iterate over potentially infinite async iterators. In this situation, using an async for loop may be inappropriate. Alternatively, you can use anext() in a while loop.

Consider the following async generator that yields potentially infinite integer numbers on demand:

import asyncio

async def async_inf_integers(start=0):
    current = start
    while True:
        yield current
        current += 1
        await asyncio.sleep(0.5)

This async generator function yields a potentially infinite stream of integer numbers. The call to asyncio.sleep() simulates an asynchronous operation here.

To process this iterator, you can use a while loop along with the anext() function instead of using an async for loop:

# ...

async def main(stop=5):
    generator = async_inf_integers()
    while True:
        number = await anext(generator)
        # Process the number here...
        print(number)
        if number == stop - 1:
            break

asyncio.run(main(20))

In this code snippet, you have a main() function that implements a potentially infinite while loop. The code explicitly communicates that you’re running a potentially infinite loop, which would be harder to communicate with an async for loop.

The anext() function lets you retrieve numbers from the async iterator on demand. Then, you can process the current number. Finally, you use the stop argument in a conditional to stop the loop.

Asynchronous Comprehensions and Generator Expressions

You can also use async iterators and iterables in asynchronous comprehensions. To create an async comprehension, you can use the following syntax:

• List comprehensions: [item async for item in async_iter]
• Set comprehensions: {item async for item in async_iter}
• Dictionary comprehensions: {key: value async for key, value in async_iter}

These comprehensions look like regular comprehensions. The only difference is that you need to use the async for construct, and the async_iter object should be an asynchronous iterator, iterable, or generator.

To illustrate how async comprehensions work, consider the following example:

import asyncio

async def async_range(start, end):
    for i in range(start, end):
        await asyncio.sleep(0.2)
        yield i

async def main():
    number_list = [i async for i in async_range(0, 5)]
    number_dict = {i: str(i) async for i in async_range(0, 5)}
    print(number_list)
    print(number_dict)

asyncio.run(main())

In the first highlighted line, you use a list comprehension to generate five integer numbers using the async_range() generator function. In the second highlighted line, you create a dictionary comprehension using the numbers as keys and their string representation as values.

When you run the example, you’ll have to wait for the code to complete, and then you’ll get the following output on your screen:

$ python async_comp.py
[0, 1, 2, 3, 4]
{0: '0', 1: '1', 2: '2', 3: '3', 4: '4'}

Both comprehensions work as expected. You can play around with other examples and generate a set of numbers, a list of squares, and so on.

Finally, you can also create asynchronous generator expressions with the following syntax:

(item async for item in async_iter)

Async generator expressions are similar to async comprehensions, but the difference is that they use parentheses instead of other brackets. In this case, instead of a list, set, or dictionary, you get an async generator iterator. Then, you can use this iterator as you’d use a regular one.