By now, you're comfortable with async and await, and you've likely used asyncio.gather to fire off a bunch of requests at once. But in the real world, you often deal with streams of data—like paginated API responses or logs—where you don't want to wait for the entire dataset to load into memory before you start processing it. That's where async generators and async comprehensions come in.

Let's build a small tool that simulates fetching paginated user data from a remote server. We want to stream these users one by one, but we only care about the users who have a "premium" status.

Simulating a paginated data stream

First, I'll create a mock function that simulates a network request. I'm using asyncio.sleep here to mimic the latency you'd actually see when hitting a REST endpoint. I'll keep it simple: the function takes a page number and returns a list of users.

import asyncio
import random

async def fetch_page(page: int):
    print(f"Fetching page {page}...")
    await asyncio.sleep(0.5)  # Simulate network lag
    # Simulate 3 users per page
    return [{"id": i, "name": f"User {i}", "premium": random.choice([True, False])} 
            for i in range(page * 3, (page + 1) * 3)]

Turning the fetcher into an async generator

Now, I don't want to call fetch_page ten times and store all those lists in a giant master list. That's a waste of RAM. Instead, I'll wrap this in an async generator. By using yield inside an async def function, I'm telling Python: "I'm going to give you values over time, and you can await each one."

async def get_user_stream(total_pages: int):
    for page in range(total_pages):
        users = await fetch_page(page)
        for user in users:
            yield user

Where I usually trip up (and you might too)

Here is where I've seen a lot of developers—including myself in the past—get frustrated. If I try to iterate over get_user_stream using a standard for loop, Python will throw a TypeError telling me that the 'async_generator' object is not iterable.

# This will FAIL
async def main():
    # TypeError: async_generator object is not iterable
    for user in get_user_stream(3):
        print(user)

The mistake is forgetting that since the generator yields values asynchronously, the loop itself must be asynchronous. I have to use async for. This tells the event loop to pause the loop and wait for the next value to be yielded by the generator before continuing.

# This is the correct way
async def main():
    async for user in get_user_stream(3):
        print(f"Processed {user['name']}")

Filtering the stream with async comprehensions

Finally, let's say we don't want to print every user; we want to collect a list of only the premium users. While we could do this with a loop and list.append(), Python gives us async comprehensions. They look and feel exactly like standard list comprehensions, but they allow you to iterate over an async generator.

I'll update my main function to use one. This is incredibly clean because it handles the async for logic behind the scenes.

async def main():
    # This is an async comprehension
    premium_users = [user async for user in get_user_stream(5) if user['premium']]
    
    print(f"Found {len(premium_users)} premium users: {[u['name'] for u in premium_users]}")

asyncio.run(main())

I personally love this pattern because it separates the concern of data retrieval (the async generator) from the concern of data filtering (the async comprehension). It keeps your memory footprint low while keeping your code readable.

📋 Practical Task

Build a Real-Time Sensor Data Filter

You are building a monitoring system for a factory. You need to process a stream of temperature readings from an asynchronous sensor.

Your task:

1. Create an async generator called read_sensor_data(limit) that yields a random float between 20.0 and 100.0 for a specified number of limit iterations. Simulate a hardware delay using await asyncio.sleep(0.1).
2. Use an async comprehension to create a list called alerts that only contains readings greater than 80.0.
3. Print the final alerts list to the console.

import asyncio
import random

# Write your code here...

async def main():
    # Implement the async comprehension here
    pass

asyncio.run(main())