Building APIs with FastAPI

Every API lesson so far has had you on the calling side of the table. This week the table turns: three teams want your 311 data, and the database is not a handoff format. You are about to build your first real service — a FastAPI app over the CityOps intake pipeline, with a contract other people can build on.

Assumes: Milestone 2 (the SQLite service_requests table), plus Pydantic models from the contract lesson. Installs: pip install fastapi "uvicorn[standard]" httpx. Everything below ran — the outputs are pasted from real runs against a 6-row test database, stated exactly.

Before you code: clarify the ask

Input: the Milestone 2 SQLite table — fingerprint PRIMARY KEY, unique_key, created_at (UTC), agency, complaint_type, descriptor, borough, incident_zip, city
Output: a running HTTP service other teams can query, with a contract
Deadline: the council dashboard team starts building next week

Notice the constraint Dev slipped in: read-only. That single word deletes half the design space — no authz model for writes, no idempotency keys on POST, no conflict resolution. Scoping is a feature.

The minimal concept

Three ideas, and you've met two of them from the other side:

A route is a Python function with a URL. @app.get("/requests") says: when someone GETs this path, run this function and send back what it returns. FastAPI handles the HTTP plumbing — parsing the request, serializing the response — and you write the function.

The Pydantic model is the contract — in both directions. In Milestone 2 the model validated data coming in. Here response_model=list[ServiceRequest] validates data going out: every response is checked against the model before it leaves, and anything the model doesn't declare never reaches the client. The model is the promise you sign.

Status codes are the vocabulary. 200 "here it is." 404 "that doesn't exist." 422 "your request was malformed — here's exactly what was wrong." 500 "we messed up." A client can only handle what you tell it; the codes are how you tell it.

Build it: the other side of the table

The whole service is one file. Read it top to bottom — there's nothing in it you haven't met before, just pointed the other way:

"""CityOps read API: the other side of the table."""
import sqlite3
from typing import Optional

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

DB_PATH = "cityops.db"


class ServiceRequest(BaseModel):
    """What a stored 311 request looks like to the outside world."""
    fingerprint: str
    unique_key: str
    created_at: str          # UTC ISO instant, stored as TEXT
    agency: str
    complaint_type: str
    descriptor: Optional[str] = None
    borough: Optional[str] = None
    incident_zip: Optional[str] = None
    city: Optional[str] = None


app = FastAPI(title="CityOps 311 API", version="0.1.0")


def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


@app.get("/requests", response_model=list[ServiceRequest])
def list_requests(
    borough: Optional[str] = Query(None, description="e.g. BROOKLYN"),
    agency: Optional[str] = Query(None, description="e.g. DSNY"),
    complaint_type: Optional[str] = Query(None),
    limit: int = Query(20, ge=1, le=100, description="page size, max 100"),
    offset: int = Query(0, ge=0, description="rows to skip"),
):
    """Filterable, paged list of stored 311 requests."""
    clauses, params = [], []
    for column, value in (("borough", borough), ("agency", agency),
                          ("complaint_type", complaint_type)):
        if value is not None:
            clauses.append(f"{column} = ?")
            params.append(value.upper() if column == "borough" else value)
    where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
    params += [limit, offset]
    with get_db() as conn:
        rows = conn.execute(
            f"SELECT * FROM service_requests {where} "
            "ORDER BY created_at LIMIT ? OFFSET ?", params).fetchall()
    return [dict(r) for r in rows]


@app.get("/requests/{fingerprint}", response_model=ServiceRequest)
def get_request(fingerprint: str):
    """One request, by its content fingerprint."""
    with get_db() as conn:
        row = conn.execute(
            "SELECT * FROM service_requests WHERE fingerprint = ?",
            (fingerprint,)).fetchone()
    if row is None:
        raise HTTPException(
            status_code=404,
            detail=f"no request with fingerprint {fingerprint!r}")
    return dict(row)

Two things to notice before we run it. First, the query parameters are declared — limit: int = Query(20, ge=1, le=100) says "an integer, default 20, between 1 and 100." FastAPI enforces that before your function ever runs. Second, the SQL uses ? placeholders, never string interpolation of user input — the filter values go in params, not in the query text.

Now the runs. Six rows seeded into a test database (stated exactly — six, not "a few"), exercised with FastAPI's TestClient — the same no-server-needed pattern as Post 7's scripted fakes:

seeded rows: 6
GET /requests -> 200 | rows: 6
GET /requests?borough=brooklyn -> 200 | fps: ['fp001', 'fp003']
GET /requests?agency=DSNY&limit=2&offset=1 -> 200 | fps: ['fp003', 'fp005']

The borough filter lowercases nothing and demands nothing — it uppercases the input, because the pipeline stores boroughs uppercase (Milestone 2's cleaning step). The client says brooklyn; the database holds BROOKLYN; the API translates. That's the contract absorbing a difference instead of leaking it.

One record, by fingerprint:

GET /requests/fp004 -> 200
{
  "fingerprint": "fp004",
  "unique_key": "10000004",
  "created_at": "2026-09-21T09:40:00+00:00",
  "agency": "DOB",
  "complaint_type": "Plumbing",
  "descriptor": "Leak",
  "borough": "MANHATTAN",
  "incident_zip": null,
  "city": "New York"
}

Note what's missing: ingested_at. The table has ten columns; the model declares nine. response_model silently drops the tenth — the client sees the contract, not the storage. When you add an internal column next month, the outside doesn't change. That's the answer to Dev's SQLite-file proposal, in one missing field.

Break it, three ways

1. The missing row. Ask for a fingerprint that doesn't exist:

GET /requests/does-not-exist -> 404
{"detail": "no request with fingerprint 'does-not-exist'"}

That's the whole break, and it's the polite one. The alternative — letting the missing row surface as an unhandled error — would be a 500 with no useful body, and a 500 means "we messed up," which is a lie: the client asked for something that isn't there. HTTPException(404) is the honest code path, and the detail string is the part Maria's teams will paste into their own debugging. Errors are a user interface.

2. The greedy client. Dev's constraint — "ten million rows in one call shouldn't take down the database" — is the le=100 on limit. Watch it work:

GET /requests?limit=500 -> 422
{
  "detail": [
    {
      "type": "less_than_equal",
      "loc": ["query", "limit"],
      "msg": "Input should be less than or equal to 100",
      "input": "500",
      "ctx": {"le": 100}
    }
  ]
}

That JSON is generated, not written — FastAPI validated the query param against the declaration and rejected it before the function ran. Same for a wrong type entirely:

GET /requests?limit=abc -> 422
detail: {'type': 'int_parsing', 'loc': ['query', 'limit'],
         'msg': 'Input should be a valid integer, ...', 'input': 'abc'}

The loc field is doing quiet, excellent work: it tells the client where the problem is (query → limit) and what was wrong. Compare that to a hand-rolled int(request.args["limit"]) raising a 500 with a traceback. The 422 says "you asked wrong, here's how to ask right" — which is the difference between a client that fixes itself and a client that emails Maria.

3. The SQLite-file "solution." Back to Dev's Monday proposal, because it deserves a proper burial. Handing teams the database file means: no contract (they read raw storage, so every schema change breaks them); no boundaries (a bad query locks the file the pipeline writes to); no evolution (you can never rename a column, because three teams' scripts say otherwise). The API costs one file and buys you all three back. The file is free; the coupling is forever.

Productionize: run it, document it, test it

Running it for real — uvicorn is the server, the app is the code:

$ uvicorn cityops_api:app --port 8123
INFO:     Started server process [7029]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8123 (Press CTRL+C to quit)

$ curl "http://127.0.0.1:8123/requests?agency=HPD&limit=1"
[{"fingerprint":"fp002","unique_key":"10000002",
  "created_at":"2026-09-20T13:22:00+00:00","agency":"HPD",
  "complaint_type":"HEAT/HOT WATER","descriptor":"No heat",
  "borough":"BRONX","incident_zip":"10451","city":"Bronx"}]

Then the docs you didn't write. FastAPI generates them from the models and declarations:

GET /docs -> 200 | content-type: text/html; charset=utf-8
GET /openapi.json -> 200 | title: CityOps 311 API

/docs is an interactive page — try the endpoints in the browser, see the parameters, the models, the error shapes. Maria's whole feature ("docs they can read without calling you") arrived free, generated from the same declarations that enforce the contract. When the code and the docs are generated from the same source, they can't drift apart.

And the testing story is Post 7's discipline applied to a service: every run above went through TestClient, which exercises the real app — real routing, real validation, real 404s and 422s — with no server process. The suite for this service asserts the contract: 200 for known fingerprints, 404 for unknown, 422 for limit=500. If someone edits the model next month, the suite catches the drift before the council dashboard team does.

Explain it to the customer

"Maria — the three teams get a URL, not a database file. /requests lists 311 records with filters for borough, agency, and complaint type, twenty at a time; /requests/{fingerprint} fetches one. There's an interactive docs page where they can try it themselves — send them that link instead of a CSV. It's read-only: they can't touch the intake tables. And Dev — the page size is capped at 100 per call, so nobody can accidentally ask for ten million rows. If a team asks for something that doesn't exist, they get a clear 404, not a crash; if they ask wrong, a 422 that says exactly what was wrong. When we change the database next month, their code keeps working — the API is the contract, and we control it."

The pattern, once more: what they get, what they can't break, and where to go instead of emailing a human.

Where this lands in CityOps

This is the read layer the whole platform grows around. The intake pipeline writes; the API reads — and everything downstream (the operator console, the exports, the AI copilot) talks to the API, never to the raw tables. That separation is what lets the inside evolve: the pipeline team can rework storage while the serving contract holds still. Later in this stage the service grows teeth — enrichment, alerts, an operator-facing front end, and Lisa's access rules — but the shape stays the one you built here: routes, models, honest errors, free docs.

The signature line for this one: an API is a contract you sign with strangers. You spent the first half of this stage learning to read contracts other people wrote. Now you've written one — and three teams you've never met are about to build on it.