Exercise py-12 — the F1 API, made fully async

In a synchronous framework, one blocking database call just makes that request slow. In an async one it stalls every other request sharing the event loop — the failure mode changes shape entirely. You extend py-10's service with a pooled connection layer (asyncpg-shaped; a FakePool stands in for Postgres in the tests) that cloud-09's RDS instance will sit behind for real, a /readyz that reflects the pool's health instead of a stub model's, and a shutdown that drains in-flight requests before it closes the pool out from under them. It runs on your Mac rather than in the browser for the same reason py-10 does: Pyodide has no FastAPI, no real ASGI server, and no connection pool to exhaust.

~90 minruns locally · uv + pytest 6 checkspy-12-async-service

What you're building

create_app(pool) — POST /ask now acquires a connection with async with pool.acquire() as conn: before it can answer, and logs the exchange with one await conn.execute(...); the request-id middleware carries over from py-10 unchanged.
GET /readyz — {"ready": pool.healthy}, 503 the moment the pool reports itself lost, exactly the signal a target group's health check needs to pull a task out of rotation before it fails real traffic.
a drain_and_close() shutdown hook — waits for every checked-out connection to come back, then closes the pool, so a deploy's rolling restart never cuts off a request mid-flight.

The tests are ordinary pytest and ship in the public folder with the starter — read them first; the names below are the check list. Solutions are not published.

Get the repo (once)

Needs git. uv installs the right Python itself, so nothing else is required.

# once, anywhere on your machine
git clone https://github.com/theDocWho/ai-ml-roadmap.git
cd ai-ml-roadmap

No git? Download the ZIP, unzip it, and cd into the unzipped folder instead.

Run it

From the repo root:

# one-time: uv (https://docs.astral.sh/uv/) manages the venv and pins Python ≥ 3.12
cd exercises/py-12-async-service && uv sync && uv run pytest -q

# once the tests pass: talk to it for real (needs a reachable Postgres — cloud-09's RDS, or any local one)
DATABASE_URL=postgresql://... uv run uvicorn f1api.main:app --port 8000 &
curl -s -X POST localhost:8000/ask -H 'content-type: application/json' -d '{"prompt": "hi"}'
curl -si localhost:8000/readyz

# load test — watch p95 climb once concurrent users pass F1API_POOL_MAX
uv run locust -f locustfile.py --host http://127.0.0.1:8000

# the same bar the reference solution clears
uv run ruff check . && uv run mypy src

Done when uv run pytest -q prints 6 passed. The untouched starter fails all six.

The checks

test_no_sync_db_calls_or_blocking_sleep_in_request_path — a static scan of src/ for psycopg2 or a blocking time.sleep, plus one real /ask call proving the pool actually got used (an empty starter can't pass a pure grep vacuously).
test_pool_size_is_respected_under_concurrent_requests — 9 concurrent /ask calls against a pool of 3 peak at exactly 3 connections checked out at once.
test_100_concurrent_requests_complete_under_the_fake_pool — 100 concurrent /ask calls against a pool of 8 all return 200.
test_graceful_shutdown_drains_inflight_requests — 4 requests already holding a connection when shutdown runs all still complete with 200; the pool only closes once they have.
test_request_ids_preserved_under_concurrent_requests — 4 concurrent requests, 4 distinct X-Request-Id headers, each response's header and body still match its own request after an await pool.acquire() lands in the middle of the handler.
test_readyz_flips_to_503_when_pool_is_lost — 200 while the pool is healthy, 503 with {"ready": false} the moment it isn't.

Files

exercises/py-12-async-service/README.md — this brief, offline
pyproject.toml — deps, ruff and mypy config
locustfile.py — the load test
src/f1api/app.py — the starter you edit
src/f1api/pool.py — FakePool, the asyncpg stand-in
tests/test_service.py — the checks

I ran the tests locally and they pass (6 passed).

This is self-attestation — the site cannot see your terminal, so the box and the button are you telling The Path the suite went green on your machine.

If you get stuck

test_pool_size_is_respected_under_concurrent_requests reads a peak of 0 — /ask never called pool.acquire() at all; check the insert is inside the async with block, not after it.
test_graceful_shutdown_drains_inflight_requests raises PoolClosedError — the shutdown handler called pool.close() without first waiting for pool.in_use to hit 0; the fake pool raises immediately instead of hanging the way the real one would.
test_request_ids_preserved_under_concurrent_requests is red — a plausible bug is reading the request id back from a module-level variable instead of current_request_id(); under real concurrency (the whole point of this test) a later request overwrites it before an earlier one reads it back.
Reading a red row — uv run pytest -q -x --tb=short stops at the first failure and shows the assertion that tripped; the message names the observed value, not the fix.