Skip to main content

Async Iteration and Context Managers

If you're new to programming

async for and async with are the async versions of for and with that you already know. They work the same way, but they let the event loop do other things while waiting for the next item or the next resource.

If you've coded before

AsyncIterator protocol, async context managers, aiofiles for non-blocking file I/O. async with aiofiles.open() drops in where with open() was. async for consumes AsyncIterator/AsyncGenerator types. This lesson completes the async toolkit for SmartNotes.

In Lesson 4, James used asyncio.gather to export five notes concurrently. The timing dropped from 1 second to 0.2 seconds. But inside each coroutine, filepath.write_text() is still a regular synchronous call. It blocks the event loop while the disk writes.

"The gather worked," James says. "But the file writes inside each coroutine are still synchronous. If one file is large, it holds up the event loop."

"You are right," Emma says. "For small files, you will not notice. But the proper tool exists: aiofiles. It wraps file I/O so the event loop stays free during reads and writes. And while we are upgrading, Python has async versions of for and with too."

async with: Non-Blocking Resources​

You know with open(...) as f: from Chapter 62. It opens a file, lets you work with it, and closes it automatically. async with does the same thing, but without blocking the event loop.

First, install aiofiles:

uv add aiofiles

Now compare the two patterns:

# Sync (blocks the event loop during file I/O)
from pathlib import Path

def read_note_sync(filepath: Path) -> str:
    with open(filepath) as f:
        return f.read()
# Async (non-blocking)
import aiofiles
from pathlib import Path

async def read_note_async(filepath: Path) -> str:
    async with aiofiles.open(filepath, mode="r") as f:
        content = await f.read()
    return content

The structure is identical: open, read, close. The only differences are async with instead of with, and await f.read() instead of f.read(). The file handle is closed automatically when the async with block ends, the same way regular with works.

Writing works the same way:

import aiofiles
from pathlib import Path


async def write_note_async(filepath: Path, content: str) -> None:
    async with aiofiles.open(filepath, mode="w") as f:
        await f.write(content)

Test it:

import asyncio
from pathlib import Path
import aiofiles


async def main() -> None:
    filepath = Path("test_note.md")

    # Write
    async with aiofiles.open(filepath, mode="w") as f:
        await f.write("# Test Note\n\nThis was written asynchronously.\n")

    # Read back
    async with aiofiles.open(filepath, mode="r") as f:
        content = await f.read()

    print(content)


asyncio.run(main())

Output:

# Test Note

This was written asynchronously.

The result is identical to synchronous file I/O. The difference is invisible to the user but matters to the event loop: while aiofiles waits for the disk, other coroutines can run.

async for: Iterating Over Async Sources​

In Chapter 64, you wrote generator functions that yield values one at a time. An async generator does the same thing, but each value may involve an await:

import asyncio
from pathlib import Path
from collections.abc import AsyncIterator
import aiofiles


async def read_notes_async(directory: Path) -> AsyncIterator[dict[str, str]]:
    """Yield note contents from all .md files in a directory.

    The return type AsyncIterator[dict[str, str]] means: this function
    produces dict items one at a time (like a generator), and you
    consume them with 'async for'. It is the async version of the
    Iterator type you used in Ch 64 for sync generators.
    """
    for filepath in sorted(directory.glob("*.md")):
        async with aiofiles.open(filepath, mode="r") as f:
            content = await f.read()
        yield {"filename": filepath.name, "content": content}

Consume it with async for:

async def main() -> None:
    directory = Path("notes")
    async for note in read_notes_async(directory):
        title = note["content"].split("\n")[0]
        print(f"{note['filename']}: {title}")


asyncio.run(main())

Output (given three .md files in notes/):

async-intro.md: # Async Intro
cli-tools.md: # CLI Tools
python-basics.md: # Python Basics

async for calls __anext__() on the async generator, which may await internally. Between yields, the event loop is free to handle other tasks. This matters when reading from a slow source like a network API or a large directory.

Regular for would not work here:

# This raises a TypeError:
for note in read_notes_async(directory):  # TypeError: 'async_generator' is not iterable
    print(note)

Async generators require async for. The types tell you: if the function has async def and yield, it returns an AsyncGenerator, and you must use async for to consume it.

SmartNotes Async File Reader​

Combine async reading with the concurrent export from Lesson 4. This creates a full async pipeline: read notes from disk, process them, export to a new format.

import asyncio
from pathlib import Path
from collections.abc import AsyncIterator
from dataclasses import dataclass
import aiofiles


@dataclass
class Note:
    title: str
    body: str
    tags: list[str]


async def read_notes_from_directory(directory: Path) -> list[Note]:
    """Read all note files from a directory asynchronously."""
    notes: list[Note] = []
    async for data in read_note_files(directory):
        lines = data["content"].strip().split("\n")
        title = lines[0].lstrip("# ").strip() if lines else "Untitled"
        body = "\n".join(lines[2:]) if len(lines) > 2 else ""
        tags_line = [line for line in lines if line.startswith("Tags:")]
        tags = (
            [t.strip() for t in tags_line[0].replace("Tags:", "").split(",")]
            if tags_line
            else []
        )
        notes.append(Note(title=title, body=body, tags=tags))
    return notes


async def read_note_files(directory: Path) -> AsyncIterator[dict[str, str]]:
    """Yield file data from all .md files in a directory."""
    for filepath in sorted(directory.glob("*.md")):
        async with aiofiles.open(filepath, mode="r") as f:
            content = await f.read()
        yield {"filename": filepath.name, "content": content}


async def export_note(note: Note, directory: Path) -> Path:
    """Export a single note using aiofiles."""
    slug = note.title.lower().replace(" ", "-")
    filepath = directory / f"{slug}.md"
    content = f"# {note.title}\n\n{note.body}\n\nTags: {', '.join(note.tags)}\n"
    async with aiofiles.open(filepath, mode="w") as f:
        await f.write(content)
    return filepath


async def async_pipeline(
    source_dir: Path, export_dir: Path
) -> int:
    """Read notes from source, export to destination concurrently."""
    notes = await read_notes_from_directory(source_dir)
    export_dir.mkdir(parents=True, exist_ok=True)
    coroutines = [export_note(note, export_dir) for note in notes]
    results: list[Path] = await asyncio.gather(*coroutines)
    return len(results)


async def main() -> None:
    count = await async_pipeline(
        source_dir=Path("notes"),
        export_dir=Path("export_output"),
    )
    print(f"Pipeline complete: {count} notes processed")


asyncio.run(main())

Output:

Pipeline complete: 5 notes processed

The pipeline has two async stages: reading (async for + aiofiles) and writing (gather + aiofiles). Both stages keep the event loop free for other work. In a web server, this pattern lets you handle file operations without blocking incoming requests.

PRIMM-AI+ Practice: Predict the Async Pipeline​

Predict [AI-FREE]​

Press Shift+Tab to enter Plan Mode.

Given this code:

import asyncio

async def count_up(label: str, n: int):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield f"{label}-{i}"

async def main() -> None:
    items: list[str] = []
    async for item in count_up("X", 3):
        items.append(item)
    print(items)

asyncio.run(main())
  1. What type does count_up return? (Hint: it has async def and yield.)
  2. What does items contain after the loop?
  3. Can you replace async for with a regular for here?

Write your answers before running.

Check your prediction
  1. AsyncGenerator[str, None]. The async def + yield combination makes it an async generator.
  2. ['X-0', 'X-1', 'X-2']
  3. No. A regular for raises TypeError: 'async_generator' object is not iterable. You must use async for.

Run​

Press Shift+Tab to exit Plan Mode.

Create the file and run it. Verify the output matches your prediction.

Investigate​

Ask Claude Code:

What is the difference between an async generator (async def + yield)
and a regular generator (def + yield)? Can I convert one to the other?
When would I use each?

The AI explains the protocol difference (__next__ vs __anext__) and when async generators matter: any time the data source involves I/O (files, network, databases).

Modify​

Add a filter_by_tag async generator that takes an AsyncGenerator of notes and a tag string, yielding only notes that contain that tag. Chain it: async for note in filter_by_tag(read_notes_async(directory), "python"):.

Make [Mastery Gate]​

Write the SmartNotes async file reader using aiofiles. In Claude Code, type /tdg:

  1. Write the stub: async def read_notes_from_directory(directory: Path) -> list[Note]
  2. Write tests: create temp directory with 3 note files, verify all 3 are read, verify empty directory returns empty list
  3. Prompt AI to implement using aiofiles and async for
  4. Run uv run ruff check, uv run pyright, uv run pytest

Try With AI​

Opening Claude Code

If Claude Code is not already running, open your terminal, navigate to your SmartNotes project folder, and type claude. If you need a refresher, Chapter 44 covers the setup.

Prompt 1: Upgrade Export to Use aiofiles​

Here is my export_note function from Lesson 4:

async def export_note(note: Note, directory: Path) -> Path:
    slug = note.title.lower().replace(" ", "-")
    filepath = directory / f"{slug}.md"
    content = f"# {note.title}\n\n{note.body}\n"
    filepath.write_text(content)
    await asyncio.sleep(0.2)
    return filepath

Replace the synchronous write_text with aiofiles so the
file I/O does not block the event loop. Keep the same
function signature and return type.

What you're learning: Upgrading sync code to async is a mechanical pattern: replace open() with aiofiles.open(), add async with, add await on read/write calls. The AI shows you the exact transformation, reinforcing that async file I/O is a drop-in replacement.

Prompt 2: Build a Streaming Export​

I want to read notes from one directory with async for
and write them to another directory with asyncio.gather,
using aiofiles for both reading and writing.

Write a function:
async def stream_and_export(
    source: Path, dest: Path
) -> int

It should:
1. Read notes with async for (non-blocking)
2. Collect them into a list
3. Export all concurrently with gather (non-blocking)
4. Return the count

Include type annotations on everything.

What you're learning: You are combining three async patterns in one function: async iteration for input, gather for fan-out, and aiofiles for non-blocking I/O. This is the architecture of real async pipelines in production Python.

Prompt 3: Async Pattern Decision Guide​

I now know four async patterns:
- await (run one coroutine)
- asyncio.gather (run many, wait for all)
- asyncio.create_task (run in background)
- async for (iterate over async source)

Give me a decision tree: given a problem description,
which pattern should I use? Include 3 SmartNotes examples
for each pattern.

What you're learning: Knowing four tools is not the same as knowing when to use each one. The AI builds a decision framework that maps problem shapes to async patterns, giving you a reusable mental model.

James watches the pipeline run. Read from one directory, export to another. Both stages async. No blocking.

"The pattern looks the same as regular file I/O," he says. "Open, read, close. Open, write, close. Just with async in front."

"That is the design goal of aiofiles," Emma says. "Async context managers confused me the first time I saw them. I thought I needed to learn a completely new API. The good news: for reading and writing files with aiofiles, the usage is identical to what you already know. async with instead of with, await f.read() instead of f.read(). That is the entire difference."

"And async for is the same story? Like a regular for loop but with async sources?"

"Exactly. Regular for pulls the next item synchronously. async for pulls the next item asynchronously, letting the event loop do other things between items. You built generators in Chapter 64. Async generators are the same idea, just with await inside."

James looks at the SmartNotes code. Read, process, export. All async. All concurrent. "One thing is missing. We have not tested any of this. How do you write pytest tests for async functions?"

"Same pytest you know, with one extra decorator. That is Lesson 6."