SmartNotes Async Capstone

If you're new to programming

This is a timed challenge: aim for 25 minutes, but take as long as you need. The goal is combining concepts, not speed. You have every piece from Lessons 1-6: async def, await, asyncio.gather, aiofiles, async for, and pytest-asyncio. No new concepts. The challenge is combining them into a working pipeline and proving it is faster than sequential.

If you've coded before

Full TDG on an async I/O pipeline. async_export_all exports to markdown, JSON, and CSV concurrently. async_import_all reads back. Timing comparison proves concurrency. Deliverables: smartnotes/async_io.py and tests/test_async_io.py.

Emma sets a timer. "Twenty-five minutes. You know asyncio.gather for concurrency, aiofiles for non-blocking I/O, async for for iteration, and pytest-asyncio for testing. Zero new concepts. This capstone proves you can combine them."

She writes the requirements on the board:

Function	Purpose	Returns
async_export_all(notes, directory, format)	Export notes to all 3 formats concurrently	Count of files written
async_import_all(directory)	Read all note files back from a directory	List of Notes

Deliverables:

File	Purpose
smartnotes/async_io.py	Both async functions with full type annotations
tests/test_async_io.py	Async tests with pytest-asyncio

Start the timer.

---

Step 1: Specify (5 minutes)

Design the async functions with types. Think about the signatures before writing any logic.

# smartnotes/async_io.py

import asyncio
from pathlib import Path
from dataclasses import dataclass

import aiofiles


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


async def export_note_md(note: Note, directory: Path) -> Path:
    """Export a single note as Markdown."""
    ...


async def export_note_json(note: Note, directory: Path) -> Path:
    """Export a single note as JSON."""
    ...


async def export_note_csv(note: Note, directory: Path) -> Path:
    """Export a single note as CSV row in a file."""
    ...


async def async_export_all(
    notes: list[Note],
    directory: Path,
    formats: list[str] | None = None,
) -> int:
    """Export all notes to specified formats concurrently.

    Args:
        notes: List of notes to export.
        directory: Base directory for exports.
        formats: List of formats ("md", "json", "csv"). Defaults to all three.

    Returns:
        Total number of files written.
    """
    ...


async def async_import_all(directory: Path) -> list[Note]:
    """Read all .md note files from a directory asynchronously.

    Args:
        directory: Directory containing note .md files.

    Returns:
        List of Note objects parsed from files.
    """
    ...

Run pyright on the stubs:

uv run pyright smartnotes/async_io.py

Fix any type issues before writing tests.

Hint: format-specific export structure

Each format gets its own subdirectory:

export_output/
├── md/
│   ├── python-basics.md
│   └── async-intro.md
├── json/
│   ├── python-basics.json
│   └── async-intro.json
└── csv/
    ├── python-basics.csv
    └── async-intro.csv

async_export_all creates subdirectories, builds a coroutine for each note-format combination, and gathers them all.

---

Step 2: Test (7 minutes)

Write async tests that define the expected behavior. These tests run before the implementation exists, so they should all fail.

Think about what to verify:

Test	What it checks
Export creates correct file count	3 notes x 3 formats = 9 files
Export with single format	3 notes x 1 format = 3 files
Export with empty list	Returns 0, no files created
Export creates subdirectories	md/, json/, csv/ directories exist
Import reads all notes	Finds all .md files in directory
Import from empty directory	Returns empty list
Round-trip: export then import	Export notes, import them back, compare
Timing: concurrent faster than threshold	5 notes export in under 1 second

Hint: test structure

# tests/test_async_io.py

import time
from pathlib import Path

import pytest

from smartnotes.async_io import Note, async_export_all, async_import_all


@pytest.fixture
def sample_notes() -> list[Note]:
    """Create sample notes for testing."""
    return [
        Note(title="Python Basics", body="Variables and types.", tags=["python"]),
        Note(title="Async Intro", body="Coroutines and await.", tags=["python", "async"]),
        Note(title="CLI Tools", body="Building with argparse.", tags=["python", "cli"]),
    ]


async def test_export_all_formats(
    sample_notes: list[Note], tmp_path: Path
) -> None:
    count = await async_export_all(sample_notes, tmp_path)
    assert count == 9  # 3 notes x 3 formats
    assert (tmp_path / "md").exists()
    assert (tmp_path / "json").exists()
    assert (tmp_path / "csv").exists()


async def test_export_single_format(
    sample_notes: list[Note], tmp_path: Path
) -> None:
    count = await async_export_all(sample_notes, tmp_path, formats=["md"])
    assert count == 3


async def test_export_empty_list(tmp_path: Path) -> None:
    count = await async_export_all([], tmp_path)
    assert count == 0


async def test_import_reads_notes(
    sample_notes: list[Note], tmp_path: Path
) -> None:
    await async_export_all(sample_notes, tmp_path, formats=["md"])
    md_dir = tmp_path / "md"
    notes = await async_import_all(md_dir)
    assert len(notes) == 3
    titles = [n.title for n in notes]
    assert "Python Basics" in titles


async def test_import_empty_directory(tmp_path: Path) -> None:
    notes = await async_import_all(tmp_path)
    assert notes == []


async def test_round_trip(
    sample_notes: list[Note], tmp_path: Path
) -> None:
    await async_export_all(sample_notes, tmp_path, formats=["md"])
    imported = await async_import_all(tmp_path / "md")
    original_titles = sorted(n.title for n in sample_notes)
    imported_titles = sorted(n.title for n in imported)
    assert original_titles == imported_titles


async def test_concurrent_timing(
    sample_notes: list[Note], tmp_path: Path
) -> None:
    # 5 notes should export concurrently in well under 1 second
    notes = sample_notes + [
        Note(title="Extra One", body="Extra body.", tags=[]),
        Note(title="Extra Two", body="More body.", tags=[]),
    ]
    start = time.perf_counter()
    await async_export_all(notes, tmp_path)
    elapsed = time.perf_counter() - start
    assert elapsed < 1.0  # Concurrent export of 5 notes at 0.2s each should take ~0.2s

Run the tests to confirm they fail:

uv run pytest tests/test_async_io.py -v

Every test should fail (the stubs return ...). If tests pass against stubs, your assertions are not checking anything real.

---

Step 3: Generate (3 minutes)

Prompt Claude Code to implement both functions:

Implement all functions in smartnotes/async_io.py so that
every test in tests/test_async_io.py passes.

Requirements:
- Use aiofiles for all file I/O
- Use asyncio.gather for concurrent exports
- Export formats: md (markdown), json, csv
- Each format gets its own subdirectory (md/, json/, csv/)
- import reads .md files using aiofiles and async for
- All functions must have complete type annotations
- Do not modify the test file

The AI sees your test expectations and generates code that satisfies them. Review the implementation before running verification.

---

Step 4: Verify (3 minutes)

Run the full verification stack:

uv run ruff check smartnotes/async_io.py
uv run pyright smartnotes/async_io.py
uv run pytest tests/test_async_io.py -v

Outcome	What to do
All GREEN	Move to Step 6
Some RED	Move to Step 5

Also run the timing comparison manually:

# timing_test.py
import asyncio
import time
from pathlib import Path
from smartnotes.async_io import Note, async_export_all


async def main() -> None:
    notes = [
        Note(f"Note {i}", f"Body for note {i}", ["test"])
        for i in range(20)
    ]

    # Concurrent
    start = time.perf_counter()
    count = await async_export_all(notes, Path("timing_output"))
    conc_time = time.perf_counter() - start

    print(f"Exported {count} files in {conc_time:.2f}s")


asyncio.run(main())

uv run python timing_test.py

Expected Output:

Exported 60 files in 0.XXs

Twenty notes across three formats: 60 files, all written concurrently.

---

Step 5: Debug (5 minutes)

Common async issues and how to fix them:

Problem	Symptom	Fix
Forgotten await	RuntimeWarning: coroutine was never awaited	Add await before the async function call
Wrong event loop	RuntimeError: This event loop is already running	Use asyncio.run() only at the top level, never inside an async function
aiofiles not installed	ModuleNotFoundError: No module named 'aiofiles'	Run uv add aiofiles
pytest-asyncio not configured	Tests pass but assertions never run	Add asyncio_mode = "auto" to [tool.pytest.ini_options]
Sync file I/O in async function	No error, but blocks the event loop	Replace Path.write_text() with async with aiofiles.open()
Type error with gather results	pyright complaints about list[Any]	Type the list: results: list[Path] = await asyncio.gather(...)

If tests fail, apply the debugging loop. Type /debug in Claude Code:

1. Read the failure message. Which test failed? What assertion?
2. Check if the function is actually async def (not def).
3. Check if await is used for every async call.
4. Re-prompt with the failure output if needed.
5. Verify again.

Hint: the most common capstone bug

Forgetting to create subdirectories before writing files:

# This fails:
filepath = directory / "md" / "note.md"
async with aiofiles.open(filepath, mode="w") as f:  # FileNotFoundError!
    await f.write(content)

# Fix: create the directory first
(directory / "md").mkdir(parents=True, exist_ok=True)

Make sure async_export_all creates format subdirectories before gathering the export coroutines.

---

Step 6: Read (2 minutes)

All tests pass. Review the generated code before calling it done.

Check these specifics:

1. Does async_export_all use asyncio.gather (not sequential awaits in a loop)?
2. Does async_import_all use aiofiles for reading (not Path.read_text())?
3. Are all function signatures typed, including return types?
4. Does the JSON export produce valid JSON (test by importing back)?
5. Does the CSV export handle commas in note titles (quoting)?
6. Are subdirectories created with parents=True, exist_ok=True?

If any check fails, fix it and re-run verification.

---

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: Review the Pipeline Architecture

Here is my complete async_io.py:

[paste the file]

Review the architecture. Are there any bottlenecks?
Could any part be made more concurrent? What happens
if the directory has 1,000 note files?

What you're learning: Code that works for 20 notes may not scale to 1,000. The AI identifies potential bottlenecks (too many open file handles, memory from gathering all results, directory listing performance) and suggests patterns like batching with asyncio.Semaphore.

Prompt 2: Add Progress Reporting

My async_export_all exports files concurrently but gives
no progress feedback. For 100 notes across 3 formats,
the user sees nothing until all 300 files are done.

Show me how to add a progress callback that reports
each completed file without slowing down the concurrency.
Use asyncio.create_task so reporting does not block exports.

What you're learning: Real-world async code needs observability. The AI shows how to combine create_task (from Lesson 4) with gather to report progress during concurrent operations. This pattern applies to any long-running async workload.

Prompt 3: Compare Your Pipeline to Production Patterns

I built an async file I/O pipeline for SmartNotes using
asyncio.gather and aiofiles. How does this compare to
how real Python applications handle async file operations?

Show me:
1. What FastAPI does for file uploads
2. What aiohttp does for file downloads
3. What real ETL pipelines do for batch processing

How close is my SmartNotes code to production patterns?

What you're learning: Your SmartNotes pipeline uses the same primitives as production Python applications. The AI connects your learning to real-world code, reinforcing that gather + aiofiles is not a tutorial pattern but a production pattern.

---

James runs the final timing comparison. Twenty notes, three formats, sixty files.

Exported 60 files in 0.15s

He compares it to the sequential version from Lesson 1. Twenty notes at three formats: sixty sequential writes would take several seconds. Concurrent: under a second.

"My warehouse went from loading six trucks in three hours to forty-five minutes," he says. "Same number of trucks, same cargo. We just opened all the docks at once instead of one at a time. That is what this code does."

Emma checks the test output. All green. She nods. "SmartNotes has a CLI, file I/O, testing, and now async concurrency. It reads and writes files without blocking." She pauses. "But it is still a local tool. The CLI works on your machine. What if someone wants to use SmartNotes from a mobile app? From a browser? From another program on a different computer?"

"They would need to call SmartNotes over the network."

"Exactly. They need a web API. You send an HTTP request, SmartNotes processes it, and sends back a response." Emma opens a new terminal. "Chapter 67 introduces FastAPI. Every endpoint you write there uses async def, the same pattern you just mastered. The async skills from this chapter are not just about file I/O. They are the foundation for every web API, every AI SDK call, and every MCP server you will build in Parts 5 and 6."