AI Developer Prerequisites

Overview

By the end of today you will have built, tested and containerised a small async FastAPI service backed by a SQL database, and you will be able to explain each piece out loud. Plan for about 10 hours including breaks; every block commits something to one Git repo, so you can stop and resume anywhere.

How to use this guide

• Each block has a 10-minute mini-lesson, then exercises, then a self-check. Read the lesson once, then spend most of the time typing.
• Type the code yourself. Pasting teaches nothing; the typos are where learning happens.
• Exercises marked Core are required. Stretch exercises are for when you finish early. If a block runs 50% over, skip its Stretch items and move on.
• Run everything. If a snippet does not behave as described, that is the exercise.
• Commit at the end of every block with the message suggested in the table.

Schedule

Take a 10-minute break after every block and a proper meal break after Block 4.

What you need installed: Python 3.11 or newer, Git, a GitHub account, and Docker Desktop (or Docker Engine plus the Compose plugin). Check with python3 --version, git --version and docker compose version.

Block 0: Setup (20 min)

One repo, one virtual environment, one folder per stage. A virtual environment is a private copy of Python plus its own installed packages, so this project's dependencies cannot clash with any other project on your machine.

mkdir ai-prereqs && cd ai-prereqs
git init
python3 -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install --upgrade pip
pip install fastapi "uvicorn[standard]" httpx sqlalchemy pytest pytest-asyncio
pip freeze > requirements.txt
mkdir drills app tests
touch app/__init__.py
printf ".venv/\n__pycache__/\n*.db\n.pytest_cache/\n" > .gitignore
git add . && git commit -m "chore: project skeleton"

You should end with this layout:

ai-prereqs/
  drills/            blocks 1-3 scratch files
  app/               blocks 4-5 FastAPI service
  tests/             pytest files
  requirements.txt
  .gitignore

Exercise 0.1 (Core): prove the venv works

1. Run which python (Windows: where python). The path must contain .venv.
2. Run deactivate, then which python again. Note the difference.
3. Re-activate and run pip list. Why is the list short?
4. Delete nothing, but answer in one sentence: why is .venv/ in .gitignore while requirements.txt is committed?

Done when: you can explain that the venv is rebuilt from requirements.txt on any machine, so the folder itself never needs sharing.

Block 1: Python fundamentals (90 min)

Goal: write small, typed, well-structured Python without looking things up. Work in drills/py_basics.py.

Mini-lesson (10 min): read, then run each snippet

• A class bundles data with the functions that act on it. __init__ sets up state, __repr__ controls how it prints, @property exposes a computed value as an attribute.
• A dataclass writes __init__, __repr__ and __eq__ for you from annotated fields. Mutable defaults need field(default_factory=...).
• A generator is a function that uses yield. It produces values one at a time, on demand, so it can stream data far larger than memory.
• Type hints do nothing at runtime. They document intent and let tools like mypy and your editor catch mistakes early.

from dataclasses import dataclass, field
from typing import Iterator


@dataclass
class Book:
    title: str
    pages: int
    tags: list[str] = field(default_factory=list)

    def __post_init__(self) -> None:
        if self.pages <= 0:
            raise ValueError("pages must be positive")


class Counter:
    def __init__(self) -> None:
        self._n = 0

    @property
    def value(self) -> int:
        return self._n

    def incr(self) -> None:
        self._n += 1

    def __repr__(self) -> str:
        return f"Counter({self._n})"


def evens(limit: int) -> Iterator[int]:
    n = 0
    while n < limit:
        yield n
        n += 2


maybe_title: str | None = None   # a value that may be missing

Try print(Book("Dune", 412)), list(evens(10)), and next(evens(10)) before moving on.

Exercise 1.1 (Core, 15 min): a validated dataclass

Create Task with title: str, done: bool = False and tags: list[str]. An empty or whitespace-only title must raise ValueError. Then deliberately write tags: list[str] = [] and read the error Python gives you. Fix it with default_factory and explain in one sentence why the error exists.

Exercise 1.2 (Core, 20 min): a class with behaviour

Create TaskList holding Task objects with these methods:

• add(task) appends a task.
• complete(title) marks the first task with that title done, and raises KeyError if none exists.
• pending, a @property returning the list of tasks not done.
• __len__ and __iter__, so len(tl) and for t in tl work.

Exercise 1.3 (Core, 20 min): generators

1. Write read_lines(path: str) -> Iterator[str] that opens a file and yields each stripped, non-empty line. Create sample.txt with blank lines to test it.
2. Write batched(items: Iterable[T], n: int) -> Iterator[list[T]] that yields lists of up to n items. Python 3.12 ships itertools.batched, but write your own.
3. Measure the laziness:

import sys
big_list = [i for i in range(10_000_000)]
big_gen = (i for i in range(10_000_000))
print(sys.getsizeof(big_list), sys.getsizeof(big_gen))

Write down both numbers and why they differ so much. This is why generators matter when you stream tokens or rows later in the roadmap.

Hint for batched: declare T = TypeVar("T") (import TypeVar from typing) above the function, so the annotation says the output items have the same type as the input items.

Exercise 1.4 (Core, 15 min): type hints with a checker

Add group_by_tag(tasks: Iterable[Task]) -> dict[str, list[Task]]. Annotate every function in the file, including -> None where nothing is returned. Then run:

pip install mypy && pip freeze > requirements.txt
mypy drills/py_basics.py

Introduce one wrong call such as Task(title=5) in a scratch file and confirm mypy flags it even though Python would run it.

Exercise 1.5 (Stretch, 10 min)

Add Task.from_dict(cls, data: dict[str, object]) -> "Task" as a @classmethod. Then try @dataclass(frozen=True) on Task and see what breaks when you set task.done = True. Use dataclasses.replace to work around it.

Self-check: save as drills/check_block1.py and run it from drills/

from py_basics import Task, TaskList, batched, read_lines, group_by_tag

assert Task("a").tags is not Task("b").tags
try:
    Task("   ")
except ValueError:
    pass
else:
    raise AssertionError("empty title should fail")

tl = TaskList()
tl.add(Task("a", tags=["x"]))
tl.add(Task("b", tags=["x", "y"]))
tl.complete("a")
assert len(tl) == 2
assert [t.title for t in tl.pending] == ["b"]
assert [t.title for t in tl] == ["a", "b"]
assert sorted(group_by_tag(tl)) == ["x", "y"]

assert list(batched([1, 2, 3, 4, 5], 2)) == [[1, 2], [3, 4], [5]]
lines = read_lines("sample.txt")
assert iter(lines) is lines          # generators are their own iterators
print("block 1 ok")

Explain out loud, without notes

• Why is a mutable default like [] rejected in a dataclass, and what does default_factory do?
• What is the difference between a function that returns a list and one that yields?
• What does str | None promise the reader, and what does mypy do with it?

Commit: feat: python fundamentals drills.

Block 2: Async Python (60 min)

Goal: explain what async def does and run many I/O calls concurrently with asyncio.gather. Work in drills/async_drills.py.

Mini-lesson (10 min)

• async def defines a coroutine function. Calling it does not run the body; it returns a coroutine object that something must await.
• await pauses the current coroutine and hands control back to the event loop, which runs other coroutines until the awaited thing finishes.
• asyncio.run(main()) starts the loop and runs one top-level coroutine.
• This is concurrency, not parallelism: one thread, many tasks taking turns whenever one is waiting on I/O. It helps with network and disk waits, and does nothing for CPU-heavy work.
• asyncio.gather(a, b, c) runs the awaitables concurrently and returns their results in the order you passed them.
• A blocking call such as time.sleep freezes the whole loop. Use await asyncio.sleep instead.

import asyncio
import time


async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)          # stands in for a network call
    return f"{name} done after {delay}s"


async def main() -> None:
    start = time.perf_counter()
    results = await asyncio.gather(fetch("a", 1), fetch("b", 1), fetch("c", 1))
    print(results, f"{time.perf_counter() - start:.1f}s")   # about 1.0s, not 3.0s


asyncio.run(main())

Exercise 2.1 (Core, 10 min): sequential versus concurrent

1. Time three await fetch(...) calls made one after another. Expect about 3 seconds.
2. Write async def fetch_all(delays: list[float]) -> list[str] using asyncio.gather and time it for [1, 1, 1]. Expect about 1 second.
3. Call fetch("x", 1) without await and read the "coroutine was never awaited" warning. Say what it means in your own words.

Exercise 2.2 (Core, 5 min): break the loop on purpose

Inside fetch, replace await asyncio.sleep(delay) with time.sleep(delay) and rerun the gather example. It now takes about 3 seconds. Explain why gather could not help, then restore the original line.

Exercise 2.3 (Core, 15 min): failures in a batch

Write a flaky(i: int) coroutine that raises RuntimeError when i is 3 and otherwise returns i * 10. Gather flaky(0) to flaky(4) two ways:

• Default: observe that the first exception propagates out of gather.
• With return_exceptions=True: get a list mixing results and exception objects. Print successes and failures separately using isinstance(r, Exception).

Write one sentence on when you would want each behaviour.

Exercise 2.4 (Core, 15 min): timeouts and a concurrency limit

1. Wrap a 3-second fetch in async with asyncio.timeout(1.5): and catch TimeoutError.
2. Write async def fetch_limited(delays: list[float], limit: int) -> list[str] that uses asyncio.Semaphore(limit) so no more than limit calls run at once. Ten 1-second calls with limit=3 should take about 4 seconds, because 10 calls in groups of 3 need 4 rounds.

This is the pattern you will use to call an LLM API for 100 inputs without hitting rate limits.

Exercise 2.5 (Stretch, 10 min)

Build a producer and consumer with asyncio.Queue: one task puts 10 items on the queue, three consumer tasks take items and "process" them with a short sleep. Shut down cleanly with queue.join().

Self-check: save as drills/check_block2.py

import asyncio
import time

from async_drills import fetch_all, fetch_limited


async def main() -> None:
    t = time.perf_counter()
    out = await fetch_all([1, 1, 1])
    assert len(out) == 3 and time.perf_counter() - t < 1.5

    t = time.perf_counter()
    await fetch_limited([1] * 10, limit=3)
    assert 3.5 < time.perf_counter() - t < 4.6
    print("block 2 ok")


asyncio.run(main())

Explain out loud, without notes

• What does calling an async def function return, and what makes it actually run?
• Why are three gathered one-second waits done in about one second?
• Name a task where async gives no speed-up, and say why.

Commit: feat: async drills.

Block 3: HTTP and JSON (60 min)

Goal: call an unreliable HTTP API safely. You will set timeouts, read status codes, retry only what is worth retrying, and parse JSON defensively. Work in drills/http_drills.py.

Set up a practice target (5 min)

Save this as drills/flaky_server.py. Do not study it yet; it is a deliberately unreliable server you will understand fully in Block 4.

import asyncio
import itertools

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()
counter = itertools.count(1)


@app.get("/ok")
async def ok() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/slow")
async def slow(seconds: float = 3) -> dict[str, float]:
    await asyncio.sleep(seconds)
    return {"slept": seconds}


@app.get("/flaky")
async def flaky() -> dict[str, int]:
    n = next(counter)
    if n % 3 != 0:                       # fails twice, then succeeds
        raise HTTPException(status_code=503, detail="try again")
    return {"attempt": n}


@app.get("/limited")
async def limited() -> JSONResponse:
    return JSONResponse(
        {"detail": "slow down"}, status_code=429, headers={"Retry-After": "1"}
    )

Run it in a second terminal (with the venv active): cd drills && uvicorn flaky_server:app --port 8001.

Mini-lesson (10 min)

• Timeouts: every outbound call needs one. Without it, a hung server hangs your program forever. In httpx, timeout=5.0 applies to connect, read, write and pool separately.
• Retries: retry only transient failures (the 429 and 5xx rows plus network errors), and only for requests that are safe to repeat (GET, PUT, DELETE, or a POST with an idempotency key).
• Backoff: wait longer after each failure, for example base * 2**attempt, plus a little random jitter so many clients do not retry in lockstep.
• JSON: response.json() parses the body into dicts, lists, strings, numbers, booleans and None. Always expect it to fail on bad input.

import httpx

with httpx.Client(timeout=5.0) as client:
    r = client.get("http://localhost:8001/ok")
    print(r.status_code, r.headers["content-type"], r.json())

Exercise 3.1 (Core, 10 min): read status codes

Call /ok, /nope and /flaky (four times in a row) and print the status code and body of each. Then call r.raise_for_status() on the failures and catch httpx.HTTPStatusError, printing e.response.status_code. Write down which codes you would retry and which you would not.

Exercise 3.2 (Core, 10 min): timeouts

Call /slow?seconds=3 with timeout=1.0 and catch httpx.TimeoutException. Then retry with timeout=5.0 and confirm it succeeds. Try httpx.Timeout(connect=1.0, read=5.0, write=5.0, pool=5.0) and say what each number controls.

Exercise 3.3 (Core, 20 min): retry with backoff

Write get_json_with_retry(url: str, *, max_attempts: int = 3, base_delay: float = 0.2) -> dict with these rules:

• Retry on 429, 500, 502, 503 and 504, and on httpx.TransportError (this includes timeouts).
• Do not retry any other 4xx; raise httpx.HTTPStatusError immediately.
• Wait base_delay * 2**attempt plus up to 0.1 seconds of random jitter between attempts. If the response has a Retry-After header, wait that many seconds instead.
• After the final attempt, raise the last error.
• Print one log line per attempt, for example attempt 2/3 -> 503.

Test it on /flaky (succeeds on the third attempt), /limited (watch it honour Retry-After), /nope (fails after one attempt) and /slow?seconds=3 with a 1-second timeout.

Exercise 3.4 (Core, 10 min): JSON in and out

1. Run json.loads('{"name": "Ada", "tags": ["a"], "score": null, "ok": true}') and print the type of every value. Note null becomes None and true becomes True.
2. Print it back with json.dumps(data, indent=2, sort_keys=True).
3. Parse '{"name": ' inside try/except json.JSONDecodeError and print a friendly message.
4. Try json.dumps(Task("x")) using your Block 1 dataclass and read the error. Fix it with dataclasses.asdict. Then try json.dumps({"when": datetime.now()}) and note that dates need converting to strings first.

Exercise 3.5 (Stretch, 10 min)

Make an async version with httpx.AsyncClient and await asyncio.sleep for backoff, then use asyncio.gather to fetch /flaky five times concurrently. Notice how the shared counter changes which attempts fail.

Self-check: save as drills/check_block3.py (server must be running)

import httpx

from http_drills import get_json_with_retry

BASE = "http://localhost:8001"

assert get_json_with_retry(f"{BASE}/ok") == {"status": "ok"}
assert get_json_with_retry(f"{BASE}/flaky")["attempt"] % 3 == 0

try:
    get_json_with_retry(f"{BASE}/nope")
except httpx.HTTPStatusError as e:
    assert e.response.status_code == 404      # must fail fast, no retries
else:
    raise AssertionError("expected a 404 error")

print("block 3 ok")

Explain out loud, without notes

• Why retry a 503 but never a 404 or 422?
• Why add exponential backoff and jitter instead of retrying immediately?
• What happens to a service if you call it with no timeout and it hangs?
• What is the difference between a connect timeout and a read timeout?

Commit: feat: resilient http client.

Block 4: FastAPI with validation, errors and tests (100 min)

Goal: build a small Notes API where bad input is rejected automatically, errors are deliberate, and a test suite proves it. Work in app/main.py and tests/test_notes.py. The notes live in memory for now; Block 5 moves them into a database.

Mini-lesson (15 min)

• FastAPI turns a decorated Python function into an HTTP endpoint. The type hints drive everything: they parse path and query parameters, validate bodies and generate interactive docs at /docs.
• A Pydantic model describes a JSON shape. If the request body does not match, FastAPI returns 422 with a list of what was wrong, and your function never runs.
• response_model controls what leaves the API, so internal fields never leak. status_code=201 sets the success code.
• Errors you choose are raised with HTTPException(status_code=..., detail=...). For errors specific to your domain, define an exception class and register a handler with @app.exception_handler.
• A plain def endpoint runs in a thread pool. An async def endpoint runs on the event loop, so it must only await and never block (Block 2, Exercise 2.2).
• TestClient calls your app directly in-process, so tests need no running server.

import itertools

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Notes API")
_ids = itertools.count(1)


class NoteIn(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    body: str = ""


class NoteOut(NoteIn):
    id: int


NOTES: dict[int, NoteOut] = {}


@app.post("/notes", response_model=NoteOut, status_code=201)
def create_note(note: NoteIn) -> NoteOut:
    new = NoteOut(id=next(_ids), **note.model_dump())
    NOTES[new.id] = new
    return new


@app.get("/notes/{note_id}", response_model=NoteOut)
def get_note(note_id: int) -> NoteOut:
    if note_id not in NOTES:
        raise HTTPException(status_code=404, detail="note not found")
    return NOTES[note_id]

Create pytest.ini in the repo root so tests can import app:

[pytest]
pythonpath = .
asyncio_mode = auto

Run the server from the repo root with uvicorn app.main:app --reload and open http://127.0.0.1:8000/docs.

Exercise 4.1 (Core, 15 min): see what you get for free

In the /docs page, create a note, then fetch it. Then trigger each of these and read the response body: an empty title (422), /notes/abc (422), /notes/99 (404). Write down which of the three errors FastAPI produced by itself and which one your own code raised.

Exercise 4.2 (Core, 25 min): complete the CRUD

Add these endpoints, returning 404 for a missing note where relevant:

• GET /notes with limit: int = Query(10, ge=1, le=100) and an optional q: str | None = None that keeps only notes whose title contains q, ignoring case.
• PUT /notes/{note_id} replaces title and body of an existing note.
• DELETE /notes/{note_id} returns 204 with an empty body.

Then try /notes?limit=0 and /notes?limit=500 and confirm both give 422 without any extra code from you.

Exercise 4.3 (Core, 20 min): custom validation and errors

1. Reject titles that are only whitespace with a @field_validator("title") on NoteIn. Confirm it returns 422.
2. Define class DuplicateTitle(Exception) and raise it from create and update when another note already has the same title, ignoring case.
3. Register @app.exception_handler(DuplicateTitle) returning JSONResponse(status_code=409, content={"error": "duplicate_title", "detail": ...}).

Exercise 4.4 (Core, 30 min): write the tests

Create tests/test_notes.py from this skeleton and add at least five more tests:

import pytest
from fastapi.testclient import TestClient

from app.main import NOTES, app

client = TestClient(app)


@pytest.fixture(autouse=True)
def clean_state() -> None:
    NOTES.clear()


def test_create_note() -> None:
    r = client.post("/notes", json={"title": "First", "body": "hello"})
    assert r.status_code == 201
    assert r.json()["title"] == "First"

Cover at least: fetch an existing note (200), fetch a missing note (404), empty title (422), whitespace title (422), duplicate title (409 and the error field), delete then fetch (204 then 404), and limit=0 (422). Read each note's id from the create response instead of assuming it is 1. Run pytest -q from the repo root.

If pytest prints a deprecation warning about httpx and TestClient, it comes from the library versions you installed. It is harmless while the tests pass.

Exercise 4.5 (Stretch, 15 min): a dependency

Add def require_key(x_api_key: str = Header(...)) -> None that raises 401 unless the header equals "secret", and protect DELETE with dependencies=[Depends(require_key)]. Add tests for the missing-key and correct-key cases.

Self-check

• pytest -q reports at least 7 passed and 0 failed.
• /docs lists all five routes with their status codes.
• This works against the running server:

curl -s -X POST localhost:8000/notes -H "content-type: application/json" -d '{"title":"hi"}'
curl -s -o /dev/null -w "%{http_code}\n" -X POST localhost:8000/notes -H "content-type: application/json" -d '{"title":""}'

The first prints a note with an id; the second prints 422.

Explain out loud, without notes

• Which part of your code makes a bad request return 422, and which makes a missing note return 404?
• What would leak if you removed response_model from an endpoint that returned a database row?
• When should an endpoint be def and when async def?
• Why does TestClient not need uvicorn running?

Commit: feat: notes api with tests.

Block 5: SQL and SQLAlchemy (90 min)

Goal: write joins by hand, see an index change a query plan, then persist the Notes API in a real database through SQLAlchemy. Part A is raw SQL in SQLite (45 min); Part B is SQLAlchemy 2.0 (45 min).

Part A mini-lesson (5 min)

• A primary key uniquely identifies a row. A foreign key points at another table's primary key, and the database refuses rows that break the link.
• An INNER JOIN keeps only rows that match on both sides. A LEFT JOIN keeps every row from the left table and fills the right side with NULL when there is no match.
• GROUP BY collapses rows into groups; HAVING filters those groups after counting, where WHERE filters rows before.
• An index is a sorted lookup structure. It turns "scan every row" into "jump to the matching rows", at the cost of extra storage and slower writes.

Save as drills/schema.sql, then load it with cd drills && sqlite3 practice.db < schema.sql. The sqlite3 command ships with macOS and most Linux; on Windows use DB Browser for SQLite, or python -m sqlite3 on Python 3.12 or newer.

CREATE TABLE users (
  id INTEGER PRIMARY KEY,
  name TEXT NOT NULL,
  email TEXT NOT NULL UNIQUE
);

CREATE TABLE notes (
  id INTEGER PRIMARY KEY,
  user_id INTEGER NOT NULL REFERENCES users(id),
  title TEXT NOT NULL,
  body TEXT NOT NULL DEFAULT '',
  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO users (name, email) VALUES
  ('Ada', '[email protected]'),
  ('Linus', '[email protected]'),
  ('Grace', '[email protected]');

INSERT INTO notes (user_id, title) VALUES
  (1, 'Engine'), (1, 'Analysis'),
  (2, 'Kernel'), (2, 'Git'), (2, 'Opinions');

Grace has no notes, which is what makes the LEFT JOIN exercise interesting.

Exercise 5.1 (Core, 20 min): joins

Write one query for each, run it, and predict the row count before you run it:

1. Every note with its author's name (INNER JOIN). Expect 5 rows.
2. Every user with the number of notes they wrote, including users with zero (LEFT JOIN plus GROUP BY). Grace must appear with 0.
3. Only users with more than one note (HAVING COUNT(*) > 1). Expect Ada and Linus.
4. Users with no notes at all, using LEFT JOIN ... WHERE n.id IS NULL. Expect Grace.
5. Change query 2 to an INNER JOIN and describe in one sentence what disappears and why.

Exercise 5.2 (Core, 15 min): see an index work

First add 200,000 rows:

WITH RECURSIVE seq(n) AS (
  SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 200000
)
INSERT INTO notes (user_id, title)
SELECT (n % 3) + 1, 'note ' || n FROM seq;

Then:

1. Run EXPLAIN QUERY PLAN SELECT * FROM notes WHERE user_id = 2; and note the word SCAN.
2. Run CREATE INDEX ix_notes_user_id ON notes(user_id); and repeat step 1. You should now see SEARCH ... USING INDEX.
3. Run .timer on and time SELECT COUNT(*) FROM notes WHERE title = 'note 150000'; before and after CREATE INDEX ix_notes_title ON notes(title);.
4. Write two sentences: why does users.email already have an index without you creating one, and what do indexes cost you on every INSERT?

Part B mini-lesson (10 min)

• An Engine holds the database connection pool. A Session is one unit of work: it tracks the objects you load or add, and commit() writes them in one transaction. rollback() throws the changes away.
• Declare tables as Python classes with Mapped[...] annotations. A relationship lets you move between related rows as ordinary attributes.
• Query with select(...), then run it with session.scalars(stmt).

from sqlalchemy import ForeignKey, String, create_engine, select
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, Session, mapped_column, relationship,
)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(200), unique=True)
    notes: Mapped[list["Note"]] = relationship(back_populates="user")


class Note(Base):
    __tablename__ = "notes"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
    title: Mapped[str] = mapped_column(String(100))
    body: Mapped[str] = mapped_column(default="")
    user: Mapped[User] = relationship(back_populates="notes")


engine = create_engine("sqlite:///orm_demo.db", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    ada = User(name="Ada", email="[email protected]")
    ada.notes.append(Note(title="Engine"))
    session.add(ada)
    session.commit()

    stmt = select(Note).join(Note.user).where(User.name == "Ada")
    for note in session.scalars(stmt):
        print(note.id, note.title)

Exercise 5.3 (Core, 15 min): the ORM in a scratch file

Run the snippet as drills/orm_demo.py with echo=True and read the SQL it prints. Add a second user with two notes. Write a select with a join and a where on the note title, and compare the printed SQL with the join you wrote by hand in 5.1. Delete orm_demo.db and rerun to confirm create_all rebuilds it.

Exercise 5.4 (Core, 25 min): put the Notes API on a database

Create app/db.py:

import os
from collections.abc import Iterator

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./app.db")
connect_args = {"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}
engine = create_engine(DATABASE_URL, connect_args=connect_args)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


def get_session() -> Iterator[Session]:
    with SessionLocal() as session:
        yield session

Then:

1. Add app/models.py with a NoteRow(Base) table holding id, title and body.
2. Give NoteOut the setting model_config = ConfigDict(from_attributes=True) so it can be built from a NoteRow.
3. Rewrite every endpoint to take session: Session = Depends(get_session) and use session.add, session.get, session.scalars(select(...)), session.delete and session.commit. Detect duplicate titles with func.lower(NoteRow.title) == title.lower().
4. Create the tables at startup with a lifespan handler that calls Base.metadata.create_all(engine).
5. Point your tests at a throwaway in-memory database by overriding the dependency:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.db import Base, get_session
from app.main import app

test_engine = create_engine(
    "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
TestSession = sessionmaker(bind=test_engine, expire_on_commit=False)


@pytest.fixture(autouse=True)
def fresh_db():
    Base.metadata.create_all(test_engine)

    def override():
        with TestSession() as s:
            yield s

    app.dependency_overrides[get_session] = override
    yield
    app.dependency_overrides.clear()
    Base.metadata.drop_all(test_engine)

Replace the old clean_state fixture with this one. Your Block 4 tests should pass unchanged apart from the import of NOTES, which you remove.

Exercise 5.5 (Core, 10 min): the N+1 query problem

In the scratch file with echo=True, loop over all users and print len(user.notes). Count how many SELECT statements ran: one for the users plus one per user. Now load with select(User).options(selectinload(User.notes)) and count again. Write one sentence on why this matters for an endpoint that lists 100 users with their notes.

Exercise 5.6 (Stretch, 15 min)

Add a UserRow table and an owner_id foreign key to notes, then GET /users/{user_id}/notes implemented with a join. Add a test, and add an index on owner_id.

Self-check

• pytest -q passes against the in-memory database.
• Create a note, stop the server, start it again, and GET /notes still returns it. That proves it is stored on disk in app.db.
• Starting the server with DATABASE_URL=sqlite:///./other.db uvicorn app.main:app creates a separate empty database. You will use this switch for Postgres in Block 6.

Explain out loud, without notes

• What rows does an INNER JOIN drop that a LEFT JOIN keeps?
• What is an index, why does it speed up reads, and when does it hurt?
• What is a Session, and what happens to your changes if you never call commit()?
• What is the N+1 problem, and how does selectinload fix it?

Commit: feat: sqlalchemy persistence.

Block 6: Git, Docker and CI (90 min)

Goal: collaborate through Git, start the whole app (API plus Postgres) with one command, and have GitHub run your tests on every push. Part A is Git (20 min), Part B is Docker and Compose (45 min), Part C is CI (25 min).

Part A: Git workflow

Mini-lesson (5 min)

• A commit is a snapshot with a message. A branch is a movable label pointing at a commit, so a feature can grow without touching main.
• The everyday loop is git switch -c feature/x, edit, git add, git commit, git push, then a pull request that CI checks before merge.
• A merge conflict happens when two branches changed the same lines. Git marks both versions in the file and waits for you to choose.

Exercise 6.1 (Core, 15 min)

1. Create an empty GitHub repository, then connect and push:

git remote add origin [email protected]:<you>/ai-prereqs.git
git branch -M main
git push -u origin main

2. Run git switch -c feature/health, add GET /health returning {"status": "ok"} plus a test, commit with feat: add health endpoint, switch back to main and run git merge feature/health.
3. Create a deliberate conflict: on branch a, change the title= of the FastAPI app to "Notes A"; on branch b, change the same line to "Notes B". Merge both into main and resolve the conflict by editing the markers by hand, then git add and git commit.
4. Run git log --oneline --graph --all and find both merges. Practise git diff, git restore <file> (discard an edit) and git stash / git stash pop (park work).

Part B: Docker and Compose

Mini-lesson (10 min)

• An image is a packaged, read-only filesystem plus a start command. A container is a running instance of an image; delete it and everything written inside it is gone unless stored in a volume.
• A Dockerfile builds an image in cached layers. Copy requirements.txt and install dependencies before copying your code, so editing code does not reinstall packages.
• Docker Compose describes several containers in one YAML file. Services reach each other by service name, so the API connects to the database at host db.

First add the Postgres driver: pip install "psycopg[binary]" && pip freeze > requirements.txt.

Dockerfile:

FROM python:3.12-slim
WORKDIR /code
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app ./app
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

.dockerignore:

.venv/
.git/
__pycache__/
*.db
.pytest_cache/

docker-compose.yml:

services:
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: notes
      POSTGRES_PASSWORD: notes
      POSTGRES_DB: notes
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U notes -d notes"]
      interval: 3s
      timeout: 3s
      retries: 10

  api:
    build: .
    environment:
      DATABASE_URL: postgresql+psycopg://notes:notes@db:5432/notes
    ports:
      - "8000:8000"
    depends_on:
      db:
        condition: service_healthy

volumes:
  pgdata:

Exercise 6.2 (Core, 15 min): build and run a single container

1. Run docker build -t notes-api . then docker run --rm -p 8000:8000 notes-api and open http://localhost:8000/docs.
2. Create a note, stop the container with Ctrl+C, run it again, and confirm the note is gone. Say why (the SQLite file lived inside the container).
3. Edit one line in app/main.py, rebuild, and find the CACHED marker on the pip install step. Then edit requirements.txt and watch that layer rebuild.

Exercise 6.3 (Core, 20 min): Compose with Postgres

1. Run docker compose up --build. Create a note through /docs.
2. Run docker compose down, then docker compose up again. The note should still exist, because the pgdata volume survived.
3. Inspect the database directly: docker compose exec db psql -U notes -d notes -c "select * from notes;" (use your own table name if different). Look at the API's output with docker compose logs api.
4. Run docker compose down -v, bring it back up, and confirm the data is gone. Say what -v removed.

Exercise 6.4 (Stretch, 10 min)

Add a USER instruction so the container does not run as root (create one with RUN useradd -m app), and add a Compose healthcheck on the api service that calls /health using Python's urllib.

Part C: Continuous integration

Mini-lesson (5 min)

CI means a clean machine runs your tests every time code changes, so "works on my machine" stops being an excuse. GitHub Actions reads workflow files from .github/workflows/. Each job runs on a fresh runner, and a failing step turns the pull request red.

.github/workflows/ci.yml:

name: ci

on:
  push:
    branches: [main]
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: pip
      - run: pip install -r requirements.txt
      - run: mypy app
      - run: pytest -q

  docker:
    runs-on: ubuntu-latest
    needs: test
    steps:
      - uses: actions/checkout@v4
      - run: docker build -t notes-api .

The action versions above are valid majors; check each action's Marketplace page for the newest one when you set up a real project. If mypy app fails on your code, fix the type errors rather than deleting the line.

Exercise 6.5 (Core, 20 min): make CI go green, red, green

1. Commit the workflow to main and push. Open the Actions tab and wait for a green check.
2. Create branch break-it, change one assertion in a test so it fails, push, and open a pull request. Watch the check turn red and read the failing log.
3. Fix the test on the same branch, push, and watch the PR turn green before you merge it.

Exercise 6.6 (Stretch, 10 min)

Add a third job that proves the whole stack boots from a clean checkout:

  compose:
    runs-on: ubuntu-latest
    needs: test
    steps:
      - uses: actions/checkout@v4
      - run: docker compose up -d --build --wait
      - run: curl --fail --retry 10 --retry-connrefused --retry-delay 2 http://localhost:8000/docs
      - run: docker compose down -v

In the repository settings, add a branch protection rule on main that requires the test check to pass before merging.

Self-check

• git log --oneline --graph --all shows at least one feature branch and a resolved merge.
• In a fresh folder, git clone <your repo> && cd ai-prereqs && docker compose up --build gives a working API with Postgres and no other setup.
• GitHub shows a green check on main and a red-then-green history on your break-it pull request.

Explain out loud, without notes

• What is the difference between an image and a container, and where does a container's data go when it is deleted?
• Why does the Dockerfile copy requirements.txt before app?
• Why can the API reach the database at db:5432, when localhost would fail inside the container?
• What does CI catch that running tests on your own laptop does not?

Commit: ci: docker compose and github actions.

Block 7: Capstone, a link-checker API (60 min)

Goal: combine all six topics in one small service. You submit a batch of URLs; the service checks them concurrently with timeouts and retries, stores the results in a database, and lets you query them. Work on a branch called feature/link-checker and merge through a pull request so CI has to pass.

Behaviour to build

Each result has url, status_code (nullable), ok (true for 2xx), elapsed_ms and error (nullable text).

Rules that tie the blocks together

• Python: the checker returns a @dataclass UrlResult, every function is type-annotated, and mypy app passes.
• Async: all URLs in a batch run through asyncio.gather, limited to 5 at a time by a Semaphore.
• HTTP: each request has a 3-second timeout and up to 3 attempts on 5xx, 429 and transport errors, with backoff (reuse Block 3, made async with a configurable base_delay). One bad URL must never fail the whole batch; it becomes a result with ok=false and an error message.
• FastAPI: use Field(min_length=1, max_length=20) on the list and HttpUrl for items, a 404 for unknown ids, and Depends to supply both the DB session and the shared httpx.AsyncClient.
• SQL: tables checks(id, created_at) and results(id, check_id FK, url, status_code, ok, elapsed_ms, error) with an index on results.check_id. Confirm with EXPLAIN or the SQL echo that the failures query uses the join and the index path.
• Docker and CI: docker compose up --build serves the API on Postgres, and the CI workflow runs mypy and pytest.

Suggested time plan

1. 5 min: sketch the two tables and the UrlResult dataclass on paper.
2. 25 min: models, the async check_url and check_many functions, then the three endpoints.
3. 20 min: tests (below).
4. 10 min: compose run, push the branch, open the PR, watch CI.

Testing without the internet

Inject the HTTP client as a dependency so tests can swap in a fake transport. httpx.MockTransport answers requests from a function you write:

import asyncio

import httpx


async def handler(request: httpx.Request) -> httpx.Response:
    await asyncio.sleep(0.3)                    # pretend the network is slow
    if request.url.host == "good.test":
        return httpx.Response(200, json={"hello": "world"})
    if request.url.host == "bad.test":
        return httpx.Response(503)
    raise httpx.ConnectTimeout("boom", request=request)


def make_test_client() -> httpx.AsyncClient:
    return httpx.AsyncClient(transport=httpx.MockTransport(handler))

Override your client dependency with app.dependency_overrides, the same way you overrode the session in Block 5. Write tests for all of these:

• A mixed batch of good.test, bad.test and slow.test URLs returns 201 and three results with the expected ok values.
• An empty list, 21 URLs, and "not a url" each return 422.
• GET /checks/999 returns 404.
• /failures returns only the failed results for the right batch, even when a second batch exists.
• Ten good.test URLs finish in under 2 seconds. Ten sequential 0.3-second calls would take 3 seconds, so this proves the checks run concurrently.
• A URL that returns 503 is attempted 3 times. Count calls inside the handler, and pass a tiny base_delay so the test stays fast.

Definition of done

• pytest -q passes and mypy app is clean.
• A fresh docker compose up --build serves POST /checks with a real URL such as https://example.com.
• The pull request shows a green CI check before you merge it.
• A short README.md gives the three commands to run, test and start the app.
• You can walk someone through one request from curl to the database row and back, naming where validation, concurrency, retry and the join each happen.

Stretch (if you finish early)

• Add GET /checks/{id}/export.csv that streams rows with a generator and StreamingResponse, so a million-row export would not need to sit in memory.
• Add a GET /checks?limit=&offset= list endpoint and a composite index that supports it.
• Return a Retry-After-aware error when a target URL answers 429.

Commit and merge: feat: capstone link checker.

Block 8: End-of-day checkpoint (20 min)

You are ready to move on when every row below scores at least 1 and no row scores 0. Score yourself honestly: 0 means you cannot do it, 1 means you can do it with the docs open, 2 means you can do it from memory. Edit the last column as you go.

If any row scores 0

Spend 60 to 90 minutes before Day 2 repeating that block's Core exercises from a blank file, without looking at your earlier solution. The roadmap's own advice is the same: revisit the matching part of your Python roadmap before moving on, because later topics assume these skills without re-teaching them.

Five-minute wrap-up

1. Push everything and make sure the CI badge or latest run is green.
2. Write three lines in your README.md under "Notes to self": the thing that confused you most, the thing that clicked, and the one exercise you will redo.
3. Run docker compose down -v so you start tomorrow from a clean slate.

Quick reference: the six blocks in one line each

• Python: dataclasses hold data, classes add behaviour, generators stream, type hints document.
• Async: await yields control; gather runs many waits at once; never block the loop.
• HTTP: retry transient failures with backoff; never retry client errors; always set a timeout.
• FastAPI: type hints give validation and docs; 422 is automatic, everything else you raise on purpose.
• SQL: joins combine tables, indexes speed reads, sessions batch writes into a transaction.
• Git, Docker, CI: branches isolate work, Compose reproduces the stack, CI proves it on a clean machine.