Python Async Generators for Streaming Data

Here is the article you requested, written for PythonSkillset.com.

Python Async Generators: The Secret Sauce for Streaming Data

Ever felt like your Python script is just... waiting? Loading a huge file, scraping a thousand web pages, or processing a live feed can feel like watching paint dry. You hit a bottleneck where your program can't do anything else until the data fully arrives.

There’s a more elegant way, and it combines two of Python’s most powerful features: generators and asynchronous programming. Let’s talk about async generators.

Think of a normal generator as a vending machine that gives you one soda at a time. You don't have to wait for the entire truck to unload; you get your drink, you drink it, and then you ask for the next one. That’s great for memory, but the machine is still slow at getting each soda from the back.

An async generator is like having a barista who can make your coffee while also taking the next customer's order. It yields items one by one, but it can await other tasks while it’s fetching the next piece of data. This is magic for any kind of streaming data—API responses, database cursors, or log files.

The Simple Mechanics

You write an async generator just like a normal one, but you use async def and yield. The magic happens because you can await inside the generator function.

Here’s the skeleton:

import asyncio

async def data_stream():
    for i in range(5):
        # Simulate a slow network call or file read
        await asyncio.sleep(1)
        yield f"Chunk {i}"

Notice the yield. It turns a normal async function into an async generator. You can’t just for loop over this. You need to use async for:

async def main():
    async for chunk in data_stream():
        print(chunk)

asyncio.run(main())

This will print "Chunk 0", wait a second, print "Chunk 1", and so on. The key point? While that await asyncio.sleep(1) is happening, your whole program isn't blocked. The event loop can do other work—like handling another client request or processing a different stream.

Why This Matters for Real Streaming

Let's get practical. Imagine you work for PythonSkillset.com and need to build a dashboard that shows real-time website traffic. You have an API that returns paginated results, and each page takes a few hundred milliseconds.

A blocking approach would fetch page 1, wait for the whole response, process it, fetch page 2, wait, process... this is painfully serial.

An async generator approach lets you overlap waiting and processing.

import httpx
import asyncio

async def fetch_pages(base_url):
    page = 1
    while True:
        url = f"{base_url}?page={page}"
        async with httpx.AsyncClient() as client:
            response = await client.get(url)
            data = response.json()
            if not data['items']:
                break
            yield data['items']
            page += 1
            await asyncio.sleep(0.1)  # Be nice to the API

Now, your main loop can consume this stream:

async def process_stream():
    async for page_data in fetch_pages("https://api.pythonskillset.com/traffic"):
        for item in page_data:
            update_dashboard(item)

The beauty is that process_stream can be one of many coroutines running in your event loop. Your web server can still reply to other requests while this stream is being processed.

The Two-Faced Iterator: aclose and athrow

Async generators are also context managers. They have an important method called aclose(). This lets you clean up resources cleanly if you need to stop a stream early.

async def infinite_log_stream():
    try:
        with open('server.log') as f:
            while True:
                line = await asyncio.to_thread(f.readline)
                if not line:
                    await asyncio.sleep(0.1)
                    continue
                yield line
    finally:
        print("Stream cleaning up")

# In your main code
async def monitor():
    agen = infinite_log_stream()
    async for line in agen:
        print(line)
        if "ERROR" in line:
            await agen.aclose()  # Stops the generator cleanly

When you call aclose(), it throws a GeneratorExit exception inside the generator, which triggers the finally block. No resource leaks, no ugly hacks.

When You Should (And Should Not) Use Them

Async generators are not a silver bullet. They shine when:

1. I/O is the bottleneck. Network requests, disk reads, database queries.
2. Data is produced over time. A live feed, a paginated API, a large file you want to process chunk by chunk.
3. You need to cancel mid-stream. Like stopping a log monitor.

Do not use them for purely CPU-bound tasks. A for loop over a list of numbers will be faster. Async generators add overhead—they are not meant for computational speed, but for concurrency and responsiveness.

The PythonSkillset Takeaway

At PythonSkillset, we believe that writing clean, efficient Python means not letting your program just sit there waiting. Async generators are the perfect tool for that. They let you write streaming code that feels like a simple loop but runs like a well-oiled concurrent machine.

Next time you are building something that slowly produces or consumes data—a file processor, a web scraper, a real-time dashboard—remember the async for loop. It might just be the simplest performance win you can code today.