Build a data API for city service requests

What you will be able to do

Build a real API over a real database, from the first request to the deployment steps. Analysts of a city’s transport team want their dashboards to read street-repair requests from one place, page by page, with daily numbers. The API must never change the data, and never run SQL that a client wrote. It must also never skip or repeat a row while a dashboard reads page after page.

Case studyPortfolio piece

Dataset

NYC 311 Service Requests, transport department (DOT), January to June 2025

Licence

NYC Open Data: no restrictions on use

Time

About 5 hours

Runs in

VS Code, Python 3.12+

Skills used

• SQLite from Python
• Read-only access
• Parameterised SQL
• Validated filters
• Keyset pagination
• Daily statistics
• Contract tests
• Postman

Download the starter project (.zip)Finished project (.zip)

Unzip the starter, open the folder in VS Code, and follow the steps below. They start with a new virtual environment.

The starter is a small FastAPI project with the parts that you built in the course. It has the settings, the error shape, the API key check, request IDs and the body limit. You add the database access, the endpoints, the tests and the documentation. The finished project is the result, for when you are stuck or want to compare. The project code is under the MIT licence, so you can publish your version.

Before you startWhat to know and to have readyAbout 300 minutes

What you need to know

This case study uses the whole course. The lessons it leans on most:

• One error shape, and rejecting unknown fields, from Validation and error responses.
• Settings, and test clients with their own settings, from Configuration and logging: Tests with their own settings.
• The reviewed contract and its test, from Keep the documentation true: Worked example: the reviewed contract.
• Running a collection with the Postman CLI, from Tests and repeatable runs: Run it from the terminal: the Postman CLI.
• SQL SELECT, WHERE, GROUP BY and ORDER BY. The page explains each query, so you can follow without an SQL course.

The page teaches two things that the course did not: Python’s sqlite3 module, and pagination.

What you need on your computer

• Python 3.12 or newer, VS Code, and curl, as in the course.
• About 60 MB of free disk space: the packages, and the database (8 MB to download, 36 MB unpacked).
• The Postman CLI for Part 8, from Tests and repeatable runs. The Postman app is optional.

Tested on macOS. Versions: Python 3.14.6, SQLite 3.50.4, FastAPI 0.142.2, pytest 9.1.1, Postman CLI 1.70.0. The Windows commands follow the course’s commands, but they are not tested. The deployment in Part 9 was not run for this case study.

The situation

Teresa Okafor leads a small analytics team that works for a city’s transport department. People call or write to the city’s 311 service about potholes, broken traffic signals, dark street lights and cracked sidewalks. Each call or message becomes a service request, and the department’s crews handle it. Every Monday, the managers review the past weeks: where requests come in, and how fast they are closed.

Today, the numbers come from a spreadsheet export. Someone downloads the requests, cleans them by hand, and emails the file. Each dashboard cleans the file again, a little differently. So two dashboards show two different numbers for the same week, and nobody knows which one is right. The file is also large, and it is old as soon as it is sent.

Teresa wants one source for every dashboard: a web API that serves the cleaned requests, with the same rules for everyone. Tomás Reyes, a developer on the team, builds it. Priya Nair builds the dashboards that will use it, and she asks the questions that a client asks.

The managers’ decision is where to send attention: which kinds of requests, and which boroughs, take too long to close. The API supports that decision. It does not make it.

The people and the team are invented. The requests are real: they come from New York City’s open data.

What you will build

You will build city-requests-api, a read-only API. Read-only means that it answers questions about the data, but no request can change the data. The dashboards need four things, so the API has four main endpoints:

1. A list of requests, with filters: the borough, the complaint type, the status and the creation time. The list comes in pages, and a dashboard follows the pages to the end.
2. One request by its ID, for a detail view.
3. The complaint types, with the number of requests of each type, for the dashboard’s menus.
4. Daily numbers: how many requests were opened each day, how many of them were closed within 14 days, and the median time to close.

Endpoint	What it returns
GET /v1/service-requests	A page of requests, oldest first, and a cursor for the next page
GET /v1/service-requests/{id}	One request, or 404 not_found
GET /v1/complaint-types	The complaint types, with counts
GET /v1/stats/daily	One row per day: opened, closed within 14 days, the share, the median days to close
GET /v1/dataset	The source, the snapshot date and the cleaning rules
GET /health, GET /ready	Liveness, and readiness (the database can be read)

Every endpoint under /v1 needs the X-API-Key header, and every error has the course’s shape. At the end you have a tested project with a reviewed contract, a Postman collection, a README and the steps to deploy it.

The data

The data is New York City’s dataset 311 Service Requests from 2020 to Present, on NYC Open Data. The city publishes every request, updated daily. NYC Open Data says: “There are no restrictions on the use of Open Data.” The city asks users to state the source, the version and any changes. This section and the project’s DATASET.md do that.

Nextia Learning pulled the requests of the transport department (agency DOT) on 2026-10-07: the 315,424 requests created from January 2024 to June 2025. That frozen copy is shared with the SQL course. This case study uses the first half of 2025 and applies nine cleaning rules. The most important ones:

• Keep January to June 2025. That is 110,618 requests.
• Remove repeated submissions. Two rows that are equal in every column except the ID are one request that was sent twice. 146 rows are removed.
• Leave out impossible times. 557 requests have a close time before their creation time. We cannot correct them with confidence, so they are left out.
• Keep no address and no free text. The pulled file has no street address. The rules also drop the city’s free-text answers.
• One way to say “not recorded”. “Unspecified”, “UNKNOWN” and empty values become NULL.
• Short, fixed codes, such as staten_island for the borough and in_progress for the status. A query string is easier to write with them. The SQL course keeps the city’s values, such as STATEN ISLAND. An API normalises them once, so that no client has to.

109,915 requests remain. They are in one SQLite database file, service_requests_2025h1.db. The script scripts/build_db.py in the project applies the rules, and DATASET.md lists all nine. The SQL course’s case study Prepare a leak-free table for city repair requests builds the full cleaning pipeline on the same file. It explains each rule. You do not need it here: you download the database ready-made.

The columns

Column	Meaning
id	The city’s ID of the request
created_at, closed_at	When the request was made and closed, New York local time, for example 2025-03-03T07:00:00
status	open, assigned, in_progress, pending or closed, as on 2026-10-07
complaint_type, descriptor	What the request is about, for example Street Condition and Pothole
borough	bronx, brooklyn, manhattan, queens or staten_island; NULL for 598 requests
location_type, community_board, council_district, incident_zip	Where, in the city’s words
channel	online, phone, mobile or other; NULL for 65,906 requests
latitude, longitude	The place, often an intersection; NULL for 16,647 requests
days_to_close	closed_at minus created_at, in days

Two facts matter for an API on this data. First, the times have no time zone: they are New York local time, as the city publishes them. Second, the status is a snapshot. A request that was pending on 2026-10-07 may be closed today.

Set up the project

The steps use VS Code and its terminal. The terminal opens in the project folder, which is where every command runs.

1. Download the starter project from the box at the top of this page. Unzip it into your projects folder, for example ~/projects. You get the folder city-requests-api.

2. In VS Code, choose File > Open Folder, and open city-requests-api. Then open a terminal: Terminal > New Terminal.

3. Make and activate a virtual environment:

   macOS

   python3 -m venv .venv
   source .venv/bin/activate

   Windows

   python -m venv .venv
   .venv\Scripts\Activate.ps1

   Linux

   python3 -m venv .venv
   source .venv/bin/activate

   The prompt starts with (.venv). If VS Code asks whether to use the new environment for this folder, choose Yes.

4. Install the exact versions of the course:

   python -m pip install -r requirements-lock.txt

5. Download the database, check it and unpack it:

   python scripts/get_db.py

   Output

   Downloading https://github.com/Nextia-Intelligence/nextia-learning-labs/raw/main/C16/case-studies/data/service_requests_2025h1.db.gz
   Checked service_requests_2025h1.db.gz: SHA-256 26492959add9... is correct
   Wrote data/service_requests_2025h1.db: 109,915 service requests

6. Run the tests:

   python -m pytest

   Output (end)

   tests/test_health.py ..                                                  [100%]
   
   ============================== 2 passed in 0.01s ===============================

What this tells us. The packages work, and the database is the file that we tested. The script compared its SHA-256 checksum, a fingerprint of the exact bytes, with the one that we published. A file that was cut short by a bad connection, or changed, has a different checksum, and the script stops. The two tests show that the starter app runs: /health answers, and an unknown path gets a 404 in the course’s error shape. The tests do not need the download. They make their own small database, as you see in Part 7.

Now start the API, and call it from a second terminal (in VS Code, the + in the terminal panel):

macOS

API_KEY=local-dev-key fastapi dev

Windows

$env:API_KEY = "local-dev-key"
fastapi dev

Linux

API_KEY=local-dev-key fastapi dev

Second terminal (on Windows: curl.exe)

curl -i http://127.0.0.1:8000/health

Output

HTTP/1.1 200 OK
date: Thu, 08 Oct 2026 00:25:12 GMT
server: uvicorn
content-length: 15
content-type: application/json
x-request-id: 05fc5a9e6b43

{"status":"ok"}

What this tells us. This is the first green check: the API runs, and each response has a request ID from the course’s middleware. Keep the server running while you work. fastapi dev restarts it each time that you save a Python file. The key local-dev-key is only for your computer. Every curl command below to /v1 sends it with -H "X-API-Key: local-dev-key".

Part 1: look at the database from Python

The API reads a SQLite database. SQLite keeps a whole database in one file and needs no database server. Python has a module for it, sqlite3, in the standard library, so there is nothing to install. Three ideas are enough for this project:

• sqlite3.connect(path) opens the file and returns a connection.
• con.execute(sql) runs one SQL statement, and returns a cursor over the result rows.
• .fetchone() takes one row from the cursor, and .fetchall() takes all of them, as tuples.

Start Python in the project folder with python. The prompt is >>>. Type the lines after each >>>:

>>> import sqlite3
>>> con = sqlite3.connect("data/service_requests_2025h1.db")
>>> con.execute("SELECT COUNT(*) FROM service_requests").fetchone()
(109915,)
>>> rows = con.execute("SELECT id, created_at, complaint_type, borough, status FROM service_requests ORDER BY created_at, id LIMIT 3").fetchall()
>>> for row in rows:
...     print(row)
...
(63580493, '2025-01-01T00:02:00', 'Traffic Signal Condition', 'queens', 'closed')
(63582137, '2025-01-01T00:05:00', 'Traffic Signal Condition', 'queens', 'closed')
(63578936, '2025-01-01T00:10:00', 'Traffic Signal Condition', 'queens', 'closed')
>>> con.execute("SELECT created_at, COUNT(*) FROM service_requests GROUP BY created_at ORDER BY COUNT(*) DESC LIMIT 3").fetchall()
[('2025-03-03T07:00:00', 34), ('2025-01-15T07:00:00', 32), ('2025-02-24T07:00:00', 30)]
>>> con.execute("SELECT value FROM meta WHERE name = 'pulled_on'").fetchone()
('2026-10-07',)
>>> con.close()

Type exit() to leave Python.

What this tells us. Each row is a tuple, in the order of the columns in the SELECT. The IDs are not in time order: request 63578936 was made after 63582137. So the API cannot sort by ID when the dashboards want time order. The second query finds a trap: 34 requests have exactly the same time, 07:00:00 on 3 March 2025. They are Staten Island potholes that were entered in one batch. In the whole table, 11,489 requests share their creation time with at least one other request. Remember this for pagination in Part 5. The table meta stores facts about the data, such as the snapshot date, so the API can tell its clients how old the data is.

Part 2: open the database read-only

The problem

A plain sqlite3.connect(path) can read and write. The API only needs to read. Suppose that a bug, or a request that tricks the code, runs DELETE or UPDATE. A writable connection does it, and every dashboard shows the damage. Good code makes that unlikely. A read-only connection makes it impossible, and that is a much stronger promise.

There is a second, smaller problem. SQLite expects that other programs may change the file at any time. So before each read, it locks the file and checks for changes. Our file never changes while the API runs, so that work is useless.

The idea: tell SQLite what we promise

SQLite accepts the database as a URI, a file address with options, such as file:/path/to/file.db?mode=ro&immutable=1:

• mode=ro (read-only): SQLite refuses every write on this connection.
• immutable=1: we promise that the file does not change while it is open, so SQLite does not lock it or check it.

The cost of immutable=1 is this: if someone replaces the file while the API runs, the API can read wrong data. So to publish new data, you replace the file and restart the API. Part 9 does exactly that.

Build it

Make the file city_api/db.py:

city_api/db.py

import sqlite3
from pathlib import Path

SCHEMA_VERSION = 1  # scripts/build_db.py writes it with PRAGMA user_version


class DatabaseUnavailable(Exception):
    """The database cannot be read now. The request may work later."""


def connect(path: Path) -> sqlite3.Connection:
    """Open the database read-only.

    mode=ro: SQLite refuses every write, even a wrong one in our code.
    immutable=1: we promise that the file never changes while the API runs,
    so SQLite does not lock it or check it for changes made by others.
    """
    if not path.is_file():
        raise DatabaseUnavailable(f"no database file at {path}")
    uri = f"{path.resolve().as_uri()}?mode=ro&immutable=1"
    try:
        con = sqlite3.connect(uri, uri=True, check_same_thread=False)
        version = con.execute("PRAGMA user_version").fetchone()[0]
    except sqlite3.Error as exc:
        raise DatabaseUnavailable(f"cannot read {path}: {exc}") from None
    if version != SCHEMA_VERSION:
        con.close()
        raise DatabaseUnavailable(
            f"{path} has schema version {version}, expected {SCHEMA_VERSION}"
        )
    con.row_factory = sqlite3.Row
    return con

Line	Why
path.resolve().as_uri()	Makes a correct file: address on every system, also with spaces in the path, or C:\ on Windows.
uri=True	Tells sqlite3 to read the options after ?.
check_same_thread=False	FastAPI may open the connection in one thread and use it in another. Each request gets its own connection, so this is safe.
PRAGMA user_version	A number that the build script stores in the file. If someone deploys a file with another table layout, the API says “not ready” instead of failing on every request.
con.row_factory = sqlite3.Row	Rows can be read by column name, row["created_at"], and turned into a dictionary with dict(row).

Check the promise in Python, in the project folder:

>>> from pathlib import Path
>>> from city_api import db
>>> con = db.connect(Path("data/service_requests_2025h1.db"))
>>> con.execute("SELECT COUNT(*) FROM service_requests").fetchone()[0]
109915
>>> con.execute("DELETE FROM service_requests")
Traceback (most recent call last):
  ...
sqlite3.OperationalError: attempt to write a readonly database
>>> db.connect(Path("data/missing.db"))
Traceback (most recent call last):
  ...
city_api.db.DatabaseUnavailable: no database file at data/missing.db

What this tells us. Reading works, and the DELETE fails before it changes one row: SQLite itself refuses it. A missing file gives our own exception, DatabaseUnavailable, which the API can turn into a clean answer.

Readiness: can the API read its data?

As in Publish safely, /health says “the process runs”, and /ready says “it can do its job”. Here the job needs the database. If the file is missing, /ready must say 503, and so must every data endpoint, without a traceback.

In city_api/errors.py, import the exception at the top:

city_api/errors.py (add to the imports)

from city_api.db import DatabaseUnavailable

Then add its handler in add_error_handlers, before the handler for Exception:

city_api/errors.py (add in add_error_handlers)

    @app.exception_handler(DatabaseUnavailable)
    async def database_unavailable(request: Request, exc: DatabaseUnavailable):
        logger.warning("database unavailable: %s id=%s", exc, request.state.request_id)
        return error_response(
            request,
            503,
            "database_unavailable",
            "The data is not available. Try again later.",
        )

The log line has the details, such as the path. The client gets only a stable code and a short message: the path of a file on the server is not the client’s business.

Add the model to city_api/models.py, after Health:

city_api/models.py (add)

class Readiness(BaseModel):
    status: Literal["ready"]

In city_api/main.py, add the import of db, and change the import of the models:

city_api/main.py (imports)

from city_api import db
from city_api.config import Settings, load_settings
from city_api.errors import add_error_handlers, error_response
from city_api.models import ErrorResponse, Health, Readiness

In create_app, after the warning about API_KEY, add a warning for a missing database. It is a warning, not an error: the API starts anyway, so /health and /ready can report the problem.

city_api/main.py (in create_app, after the API_KEY warning)

    if not settings.database_path.is_file():
        logger.warning("no database at %s: /ready answers 503", settings.database_path)

Then add /ready, after /health:

city_api/main.py (in create_app, after /health)

    @app.get(
        "/ready",
        tags=["health"],
        summary="Check that the service can read the data",
        responses={503: {"model": ErrorResponse, "description": "Not ready."}},
    )
    def ready() -> Readiness:
        con = db.connect(settings.database_path)
        try:
            con.execute("SELECT id FROM service_requests LIMIT 1").fetchone()
        finally:
            con.close()
        return Readiness(status="ready")

The database path comes from the setting DATABASE_PATH in city_api/config.py, with the default data/service_requests_2025h1.db. Check both cases. First, with the database:

Second terminal (on Windows: curl.exe)

curl -s http://127.0.0.1:8000/ready

Output

{"status":"ready"}

Then stop the server, and start it again with a path that does not exist:

macOS

API_KEY=local-dev-key DATABASE_PATH=data/nothing.db fastapi dev

Windows

$env:DATABASE_PATH = "data/nothing.db"
fastapi dev

Linux

API_KEY=local-dev-key DATABASE_PATH=data/nothing.db fastapi dev

Call /ready again from the second terminal:

Output of curl -s http://127.0.0.1:8000/ready

{"error":{"code":"database_unavailable","message":"The data is not available. Try again later.","request_id":"623f2048eda4","fields":[]}}

First terminal (the server's log)

WARNING city_api: no database at data/nothing.db: /ready answers 503
WARNING city_api: database unavailable: no database file at data/nothing.db id=623f2048eda4

What this tells us. The service knows whether it can do its job, and says so in the standard shape, with status 503. A hosting platform, or a monitor, can now tell “the process is down” from “the data is missing”. The log has the details and the same request ID, so a support person can find the cause. Stop the server, and start it again without DATABASE_PATH. On Windows, remove the setting first: Remove-Item Env:DATABASE_PATH.

Part 3: parameterised SQL

The problem: SQL made from text

The dashboards filter by complaint type. The obvious way to filter is to put the client’s text into the SQL, with an f-string. Try it in Python, with a read-only connection:

>>> import sqlite3
>>> con = sqlite3.connect("file:data/service_requests_2025h1.db?mode=ro", uri=True)
>>> wanted = "Bench"
>>> sql = f"SELECT COUNT(*) FROM service_requests WHERE complaint_type = '{wanted}'"
>>> con.execute(sql).fetchone()
(48,)

It works. Now suppose that a client sends this “complaint type”:

>>> wanted = "x' OR '1'='1"
>>> sql = f"SELECT COUNT(*) FROM service_requests WHERE complaint_type = '{wanted}'"
>>> print(sql)
SELECT COUNT(*) FROM service_requests WHERE complaint_type = 'x' OR '1'='1'
>>> con.execute(sql).fetchone()
(109915,)

What this tells us. The client’s text changed the question. The quote in the text closed our string, and OR '1'='1' made the condition true for every row. This is SQL injection: data from outside becomes part of the SQL code. Here it only counted all rows. With other SQL, it can read tables that the client must not see, or, on a writable connection, change and delete data. It is one of the most common security faults in real software.

Injection is not only an attack. An honest value breaks the same code:

>>> wanted = "Driver's complaint"
>>> con.execute(f"SELECT COUNT(*) FROM service_requests WHERE complaint_type = '{wanted}'").fetchone()
Traceback (most recent call last):
  ...
sqlite3.OperationalError: near "s": syntax error

The idea: values travel separately

Give SQLite the SQL with a ? where each value goes, and the values in a separate tuple. SQLite reads the SQL first, and only then uses the values, always as values, never as code:

>>> con.execute("SELECT COUNT(*) FROM service_requests WHERE complaint_type = ?", ("x' OR '1'='1",)).fetchone()
(0,)
>>> con.execute("SELECT COUNT(*) FROM service_requests WHERE complaint_type = ?", ("Bench",)).fetchone()
(48,)

What this tells us. With the ?, the attack is only a strange complaint type that no request has: zero rows. A real value still works. This is a parameterised query, and the ? is a parameter. Note the comma in ("Bench",): it makes a tuple of one value.

A parameter can stand only for a value. It cannot stand for a column name, a table name or a word such as ASC. So the rule of this project is:

• Values from a request reach SQLite only as parameters.
• Column names and SQL words come only from our code, never from a request.

Build it: one request by its ID

All the SQL of the project goes in one file, so a reviewer can check the rule in one place. Make city_api/queries.py:

city_api/queries.py

"""Every SQL statement of the API.

Values from the request reach SQLite only as parameters (the ? marks), never
as part of the SQL text. Column names and SQL words come only from this file.
"""

import sqlite3
import statistics
from datetime import date, timedelta

COLUMNS = (
    "id, created_at, closed_at, status, complaint_type, descriptor, location_type, "
    "borough, community_board, council_district, incident_zip, channel, "
    "latitude, longitude, days_to_close"
)


def get_request(con: sqlite3.Connection, request_id: int) -> sqlite3.Row | None:
    return con.execute(
        f"SELECT {COLUMNS} FROM service_requests WHERE id = ?", (request_id,)
    ).fetchone()

The f-string here is safe: COLUMNS is our own constant. The ID still goes as a parameter. (statistics, date and timedelta are for Part 6.)

The route needs an error that it raises on purpose, with a status and a code: 404 not_found for an unknown ID. Parts 4 to 6 need more of them. Add one class for all of them to city_api/errors.py, after CODES:

city_api/errors.py (add after CODES)

class ApiError(Exception):
    """An error that a route raises on purpose, with its status and code."""

    def __init__(self, status: int, code: str, message: str, fields=()):
        super().__init__(message)
        self.status, self.code, self.message, self.fields = status, code, message, fields

Its handler goes in add_error_handlers, after the handler for HTTPException:

city_api/errors.py (add in add_error_handlers)

    @app.exception_handler(ApiError)
    async def api_error(request: Request, exc: ApiError):
        return error_response(request, exc.status, exc.code, exc.message, exc.fields)

The response model describes one request. Add it to city_api/models.py, after the imports. The two Literal types list the only values that a borough and a status can have. EXAMPLE_REQUEST is a real request, which the documentation shows and a test checks:

city_api/models.py (add, after the imports)

Borough = Literal["bronx", "brooklyn", "manhattan", "queens", "staten_island"]
Status = Literal["open", "assigned", "in_progress", "pending", "closed"]

EXAMPLE_REQUEST = {
    "id": 64246800,
    "created_at": "2025-03-03T17:55:20",
    "closed_at": "2025-03-04T17:14:24",
    "status": "closed",
    "complaint_type": "Street Sign - Missing",
    "descriptor": "Speed Limit",
    "location_type": "Street",
    "borough": "manhattan",
    "community_board": "01 MANHATTAN",
    "council_district": 1,
    "incident_zip": "10013",
    "channel": "phone",
    "latitude": 40.721709619645075,
    "longitude": -74.01136041681767,
    "days_to_close": 0.97,
}


class ServiceRequest(BaseModel):
    """One request to the city about a street, a sign, a light or a sidewalk."""

    model_config = ConfigDict(json_schema_extra={"examples": [EXAMPLE_REQUEST]})

    id: int = Field(description="The city's ID of the request (unique_key).")
    created_at: str = Field(description="When it was made, New York local time.")
    closed_at: str | None = Field(description="When it was closed, as published.")
    status: Status | None = Field(
        description="Status on the snapshot date. To find closed requests, use closed_at."
    )
    complaint_type: str
    descriptor: str | None = Field(description="Detail of the complaint, such as Pothole.")
    location_type: str | None
    borough: Borough | None = Field(description="null: the city did not record it.")
    community_board: str | None
    council_district: int | None
    incident_zip: str | None
    channel: Literal["online", "phone", "mobile", "other"] | None = Field(
        description="How the request was made. null: unknown."
    )
    latitude: float | None
    longitude: float | None
    days_to_close: float | None = Field(
        description="Days from created_at to closed_at, when closed_at is set."
    )

Now the routes. Make city_api/routes.py:

city_api/routes.py

import sqlite3
from collections.abc import Iterator
from typing import Annotated

from fastapi import APIRouter, Depends, Path, Request

from city_api import db, queries
from city_api.errors import ApiError
from city_api.models import ErrorResponse, ServiceRequest
from city_api.security import require_api_key

ERRORS = {
    401: {"model": ErrorResponse, "description": "The API key is missing or wrong."},
    422: {"model": ErrorResponse, "description": "A parameter is not valid."},
    503: {"model": ErrorResponse, "description": "The data is not available."},
}

router = APIRouter(prefix="/v1", dependencies=[Depends(require_api_key)], responses=ERRORS)


def get_db(request: Request) -> Iterator[sqlite3.Connection]:
    """Give the route a read-only connection, and close it after the response."""
    con = db.connect(request.app.state.settings.database_path)
    try:
        yield con
    finally:
        con.close()


Database = Annotated[sqlite3.Connection, Depends(get_db)]


@router.get(
    "/service-requests/{request_id}",
    tags=["service requests"],
    summary="Get one service request",
    responses={404: {"model": ErrorResponse, "description": "No request has this ID."}},
)
def get_service_request(
    request_id: Annotated[int, Path(ge=1, description="The request's id.")], con: Database
) -> ServiceRequest:
    row = queries.get_request(con, request_id)
    if row is None:
        raise ApiError(404, "not_found", f"No service request has the ID {request_id}.")
    return ServiceRequest(**dict(row))

Part	Why
dependencies=[Depends(require_api_key)] on the router	Every route of the router needs the key, also the routes that you add later. You cannot forget it on one.
responses=ERRORS	Documents the three errors that every data route can give, once.
get_db with yield	A dependency, as in Separate concerns. The code before yield runs before the route, and the code after it runs after the response, even after an error. So every connection is closed.
Database = Annotated[...]	A short name for “a connection from get_db”, so each route writes con: Database.
Path(ge=1)	An ID is a whole number from 1. abc or 0 gives 422, before any SQL runs.

Opening a connection for each request costs well under a millisecond with immutable=1, and it keeps requests apart. Finally, connect the router to the app in city_api/main.py. Add the import:

city_api/main.py (add to the imports)

from city_api.routes import router

Then include the router in create_app, right after add_error_handlers(app):

city_api/main.py (in create_app)

    app.include_router(router)

Call it from the second terminal, with the key:

Second terminal (on Windows: curl.exe)

curl -s -H "X-API-Key: local-dev-key" http://127.0.0.1:8000/v1/service-requests/64246800
curl -s -H "X-API-Key: local-dev-key" http://127.0.0.1:8000/v1/service-requests/1

Output

{"id":64246800,"created_at":"2025-03-03T17:55:20","closed_at":"2025-03-04T17:14:24","status":"closed","complaint_type":"Street Sign - Missing","descriptor":"Speed Limit","location_type":"Street","borough":"manhattan","community_board":"01 MANHATTAN","council_district":1,"incident_zip":"10013","channel":"phone","latitude":40.721709619645075,"longitude":-74.01136041681767,"days_to_close":0.97}
{"error":{"code":"not_found","message":"No service request has the ID 1.","request_id":"5d784c7f4050","fields":[]}}

What this tells us. Request 64246800 is a missing speed-limit sign in lower Manhattan, reported by phone on 3 March 2025 and closed 0.97 days later. An unknown ID gives 404 with the code not_found, in the same shape as every other error. The dashboard’s detail view can show “not found” for that code, without reading the message.

Part 4: filters that cannot be misread

The problem: a typo that returns the whole city

Priya’s dashboard asks for ?borough=Bronx. The database stores bronx. A simple API passes the text to SQL, finds no rows, and returns an empty list. The dashboard shows “no requests in the Bronx”, which is false, and nobody notices. Or the dashboard writes ?boro=bronx. A simple API ignores the unknown parameter, and returns requests from every borough under the title “Bronx”.

Both mistakes look like valid answers. That is worse than an error, because an error is noticed. So the API must check every parameter, and reject what it does not know. The course did the same for unknown fields in a body, in Validation and error responses: Reject unknown fields.

The idea: a model for the query string

FastAPI can read the query parameters into a Pydantic model, as it reads a JSON body. The model gives each parameter a type, limits and a description, and extra="forbid" rejects unknown names. The models need dates and times, so change the first lines of city_api/models.py:

city_api/models.py (imports)

from datetime import date
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, NaiveDatetime

Then add the filter models after EXAMPLE_REQUEST:

city_api/models.py (add, after EXAMPLE_REQUEST)

class RequestFilters(BaseModel):
    """The filters that the list and the statistics share."""

    model_config = ConfigDict(extra="forbid")

    borough: Borough | None = Field(
        default=None, description="Only requests in this borough."
    )
    complaint_type: str | None = Field(
        default=None,
        min_length=1,
        max_length=60,
        description="Only this complaint type, exactly as in /v1/complaint-types.",
    )


class ListQuery(RequestFilters):
    status: Status | None = Field(default=None, description="Only this status.")
    created_from: NaiveDatetime | None = Field(
        default=None,
        description="Created at or after this time (New York local time, no time zone). "
        "A date such as 2025-03-01 means its first second.",
    )
    created_to: NaiveDatetime | None = Field(
        default=None,
        description="Created before this time (not included).",
    )
    limit: int = Field(default=50, ge=1, le=200, description="Requests per page.")

Choice	Why
Borough and Status are fixed lists	Bronx or unspecified gives 422 with fields: ["borough"], not an empty list.
NaiveDatetime	A time without a time zone, like the data. 2025-03-03T07:00:00Z (UTC) gives 422. Comparing a UTC time with New York times would be wrong by four or five hours, silently.
created_from included, created_to not	Two ranges that touch, such as March and April, never count a request twice.
limit from 1 to 200	The client chooses the page size, but cannot ask for the whole table at once. Part 5 explains why.
extra="forbid"	?boro=bronx gives 422 with fields: ["boro"].

Add the response model for a page after ServiceRequest. It has a next_cursor, which stays null until Part 5:

city_api/models.py (add, after ServiceRequest)

class ServiceRequestPage(BaseModel):
    """One page of requests, oldest first."""

    model_config = ConfigDict(
        json_schema_extra={
            "examples": [
                {
                    "items": [EXAMPLE_REQUEST],
                    "next_cursor": "eyJjIjoiMjAyNS0wMy0wM1QxNzo1NToyMCIsImkiOjY0MjQ2ODAwLCJmIjoiZDVmMjc4NTIifQ",
                }
            ]
        }
    )

    items: list[ServiceRequest]
    next_cursor: str | None = Field(
        description="Send it as cursor to get the next page. null: this is the last page."
    )

The SQL: conditions from a fixed table

A filter that the client did not send must not appear in the SQL. So the WHERE clause is built from the filters that are set. The parts of the SQL come from a fixed table in our code, and the values go as parameters. Add this to city_api/queries.py, after COLUMNS:

city_api/queries.py (add, after COLUMNS)

# The filters that a request may use, and the condition each one adds.
# A filter that is not in this table cannot reach the SQL.
CONDITIONS = {
    "borough": "borough = ?",
    "complaint_type": "complaint_type = ?",
    "status": "status = ?",
    "created_from": "created_at >= ?",
    "created_to": "created_at < ?",
}


def where(filters: dict) -> tuple[str, list]:
    """Build the WHERE clause and its parameters from the filters that are set."""
    parts, params = [], []
    for name, value in filters.items():
        if value is not None:
            parts.append(CONDITIONS[name])
            params.append(value)
    return (" AND ".join(parts) or "1 = 1"), params

Here is a small example. For {"borough": "bronx", "status": None, "created_from": "2025-03-01T00:00:00"}, where returns "borough = ? AND created_at >= ?" and ["bronx", "2025-03-01T00:00:00"]. With no filter, it returns "1 = 1", a condition that is always true. The times can be compared as text, because the format 2025-03-01T07:00:00 sorts in time order.

For now, the list returns the first page only. Add this function after get_request. Part 5 replaces it:

city_api/queries.py (add; replaced in Part 5)

def list_requests(con: sqlite3.Connection, filters: dict, limit: int) -> list[sqlite3.Row]:
    condition, params = where(filters)
    sql = (
        f"SELECT {COLUMNS} FROM service_requests WHERE {condition} "
        "ORDER BY created_at, id LIMIT ?"
    )
    return con.execute(sql, [*params, limit]).fetchall()

The route

In city_api/routes.py, add from datetime import datetime after from collections.abc import Iterator. Then change two import lines:

city_api/routes.py (change two imports)

from fastapi import APIRouter, Depends, Path, Query, Request

from city_api.models import ErrorResponse, ListQuery, ServiceRequest, ServiceRequestPage

Add two helpers after Database, and the list route above get_service_request, so the documentation lists it first:

city_api/routes.py (add, after Database)

def iso(moment: datetime | None) -> str | None:
    """2025-03-01 00:00 -> '2025-03-01T00:00:00', the format of the database."""
    return moment.isoformat() if moment else None


def check_range(start, end) -> None:
    if start is not None and end is not None and start >= end:
        raise ApiError(
            422,
            "invalid_range",
            "created_from must be before created_to.",
            ["created_from", "created_to"],
        )


@router.get(
    "/service-requests",
    tags=["service requests"],
    summary="List service requests, one page at a time",
)
def list_service_requests(
    query: Annotated[ListQuery, Query()], con: Database
) -> ServiceRequestPage:
    check_range(query.created_from, query.created_to)
    filters = {
        "borough": query.borough,
        "complaint_type": query.complaint_type,
        "status": query.status,
        "created_from": iso(query.created_from),
        "created_to": iso(query.created_to),
    }
    rows = queries.list_requests(con, filters, query.limit)
    return ServiceRequestPage(items=[dict(row) for row in rows], next_cursor=None)

Annotated[ListQuery, Query()] tells FastAPI to fill the model from the query string. check_range rejects a range that is empty or reversed with its own code, invalid_range. Each date alone is valid, and only the pair is wrong, so a separate code helps the client.

Try a filter and three mistakes:

Second terminal (on Windows: curl.exe)

curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?borough=staten_island&created_from=2025-03-03T07:00:00&limit=2"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?borough=Queens"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?boro=queens"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?created_from=2025-03-05&created_to=2025-03-01"

Output

{"items":[{"id":64250466,"created_at":"2025-03-03T07:00:00","closed_at":"2025-03-04T11:00:00","status":"closed","complaint_type":"Street Condition","descriptor":"Pothole","location_type":null,"borough":"staten_island","community_board":"03 STATEN ISLAND","council_district":51,"incident_zip":"10312","channel":null,"latitude":40.527655151030274,"longitude":-74.16543676128964,"days_to_close":1.17},{"id":64250471,"created_at":"2025-03-03T07:00:00","closed_at":"2025-03-04T10:30:00","status":"closed","complaint_type":"Street Condition","descriptor":"Pothole","location_type":null,"borough":"staten_island","community_board":"03 STATEN ISLAND","council_district":null,"incident_zip":"10312","channel":null,"latitude":null,"longitude":null,"days_to_close":1.15}],"next_cursor":null}
{"error":{"code":"invalid_request","message":"The request is not valid.","request_id":"cced4a993526","fields":["borough"]}}
{"error":{"code":"invalid_request","message":"The request is not valid.","request_id":"e289414ce95c","fields":["boro"]}}
{"error":{"code":"invalid_range","message":"created_from must be before created_to.","request_id":"b4c504655829","fields":["created_from","created_to"]}}

What this tells us. The filter works: two potholes from the 07:00:00 batch, in time order, then ID order. The second one has no coordinates and no council district. Real data has gaps, and the API shows them as null instead of hiding the row. Each mistake now gives 422 and names the parameter, so Priya finds her bug in seconds. But look at next_cursor: it is null, and there are more Staten Island requests after these two. The client has no way to get them yet.

Part 5: pages that never skip or repeat

The problem: too much for one answer

A dashboard that shows all requests of the half year needs 109,915 rows. In JSON, 200 requests take about 76 KB, so all of them take about 42 MB, in one response. The server builds it in memory, the network carries it, and the browser parses it, while the user waits. A second dashboard asks for the same, and then a third. So the API sends the data in pages: a limited number of rows per response, and a way to ask for the next page.

The first idea, and why it fails: OFFSET

SQL can skip rows: LIMIT 50 OFFSET 100 means “skip 100 rows, then give 50”. Page 3 is OFFSET 100. It is simple, and many APIs use it. It has two problems.

Problem 1: the data changes between two pages. A dashboard reads the open requests, two per page. Between page 1 and page 2, the city closes request 102. The starter has a small script that shows what happens, with six requests in a database in memory:

python scripts/offset_problem.py

Output

With OFFSET:
  page 1: [101, 102]
  request 102 is closed
  page 2: [104, 105]
With a keyset (after the last row of the page):
  page 1: [101, 102]
  request 102 is closed
  page 2: [103, 104]

What this tells us. With OFFSET 2, page 2 skips request 103. Request 102 left the list of open requests, so every later row moved up one place. Request 103 moved into the place that page 1 already covered. Nobody gets an error: the dashboard just never sees request 103. If a request had been added in front instead, one row would appear twice. Our database file does not change while the API runs. But the team plans a daily refresh, and a dashboard that pages during the switch has this problem.

Problem 2: deep pages get slower. To skip 100,000 rows, SQLite must still walk through them. The second script measures a page of 50 rows at different depths, on the real database:

python scripts/time_pages.py

Output (the times change a little on each run)

page after       0 rows: OFFSET  0.05 ms   keyset  0.06 ms
page after  10,000 rows: OFFSET  0.17 ms   keyset  0.06 ms
page after  50,000 rows: OFFSET  0.64 ms   keyset  0.06 ms
page after 100,000 rows: OFFSET  1.47 ms   keyset  0.06 ms

What this tells us. With OFFSET, the time grows with the depth of the page: 25 times slower at 100,000 rows. With the keyset, it stays flat. On this file, 1.5 ms is still fast, so the first problem is the real reason here. But the city has millions of requests a year. On a table 30 times larger, the last page would take about 45 ms, and the cost grows with every new row.

The idea: remember where you stopped

Think of a bookmark. You do not write “I read 100 pages”. You put the bookmark after the last page that you read. If someone removes a page before it, the bookmark is still in the right place.

A keyset is that bookmark for rows. The API sorts by (created_at, id), and remembers the last row of the page. The next page asks for the rows after that row:

WHERE (created_at, id) > ('2025-03-03T07:00:00', 64250471)
ORDER BY created_at, id
LIMIT 51

Why both columns? Remember the 34 requests at exactly 07:00:00 in Part 1. Suppose a page ends inside that group. With created_at > '2025-03-03T07:00:00' alone, the next page skips the rest of the group. With created_at >=, it repeats the group. The id breaks the tie, because no two requests have the same ID. So the pair (created_at, id) is unique, and it gives one exact place in the order. SQLite compares the pairs like words in a dictionary: first created_at, and id only when the times are equal.

Why LIMIT 51 for a page of 50? If 51 rows come back, at least one more page exists. The API returns 50, and makes a cursor. If 50 or fewer come back, this is the last page, and next_cursor is null. So the client never gets an empty last page.

The schema has an index on (created_at, id), and one on each filter column followed by (created_at, id). With them, SQLite jumps straight to the place after the bookmark. That is why the keyset time stays flat.

The cursor: opaque, and tied to its filters

The client must send the bookmark back. The API gives it as one text value, the cursor. It is opaque: the client sends it back as it is, and does not read it or build it. Then the API can change what is inside later without breaking any client.

Our cursor is a small JSON object in base64. Base64 writes any data with only letters, digits, - and _, so the cursor is safe in a URL. Inside are three values: the last created_at, the last id, and a short fingerprint of the filters. The fingerprint catches a common bug. Suppose that a dashboard takes a cursor from a list of Bronx requests and sends it with borough=queens. The bookmark belongs to a different list, so the result would be meaningless. The API refuses it with invalid_cursor.

Opaque does not mean secret. Anyone can decode base64, and that is fine. The worst that a changed cursor can do is start the list at a different place. The API checks every value in it, and the values reach SQL only as parameters.

Make city_api/pagination.py:

city_api/pagination.py

import base64
import binascii
import hashlib
import json
from datetime import datetime

from city_api.errors import ApiError


def fingerprint(filters: dict) -> str:
    """A short code for a set of filters, so that a cursor works only with them."""
    text = json.dumps(filters, sort_keys=True, default=str)
    return hashlib.sha256(text.encode()).hexdigest()[:8]


def encode_cursor(created_at: str, request_id: int, filters: dict) -> str:
    """The position after the last request of a page, as an opaque text."""
    data = {"c": created_at, "i": request_id, "f": fingerprint(filters)}
    raw = json.dumps(data, separators=(",", ":")).encode()
    return base64.urlsafe_b64encode(raw).decode().rstrip("=")


def decode_cursor(cursor: str, filters: dict) -> tuple[str, int]:
    """Return (created_at, id) from a cursor, or raise ApiError invalid_cursor."""
    try:
        raw = base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4))
        data = json.loads(raw)
        created_at, request_id = data["c"], data["i"]
        datetime.fromisoformat(created_at)
        if type(request_id) is not int:
            raise ValueError("id is not an integer")
    except (binascii.Error, UnicodeDecodeError, ValueError, KeyError, TypeError):
        raise ApiError(
            422, "invalid_cursor", "The cursor is not valid.", ["cursor"]
        ) from None
    if data.get("f") != fingerprint(filters):
        raise ApiError(
            422,
            "invalid_cursor",
            "The cursor belongs to other filters. Start again without a cursor.",
            ["cursor"],
        )
    return created_at, request_id

decode_cursor checks everything that can be wrong. The text may not be base64, or not JSON, or have no c or i. The time may not be a time, or the ID not a whole number. Each one gives the same clear 422, never a 500. The rstrip("=") removes the padding that base64 adds at the end, and decode_cursor puts it back.

Now use the cursor. In city_api/models.py, add the parameter at the end of ListQuery:

city_api/models.py (add to ListQuery)

    cursor: str | None = Field(
        default=None,
        max_length=200,
        description="next_cursor from the previous page. Use it with the same filters.",
    )

In city_api/queries.py, replace list_requests with this version:

city_api/queries.py (replace list_requests)

def list_requests(
    con: sqlite3.Connection,
    filters: dict,
    after: tuple[str, int] | None,
    limit: int,
) -> list[sqlite3.Row]:
    """One page, oldest first. Asks for limit + 1 rows to know if more exist."""
    condition, params = where(filters)
    if after is not None:
        condition += " AND (created_at, id) > (?, ?)"
        params += list(after)
    sql = (
        f"SELECT {COLUMNS} FROM service_requests WHERE {condition} "
        "ORDER BY created_at, id LIMIT ?"
    )
    return con.execute(sql, [*params, limit + 1]).fetchall()

In city_api/routes.py, import the two cursor functions, after the import of the models:

city_api/routes.py (add to the imports)

from city_api.pagination import decode_cursor, encode_cursor

Then replace the list route with the final version. It adds a description for the documentation, and the last six lines of the function:

city_api/routes.py (replace the list route)

@router.get(
    "/service-requests",
    tags=["service requests"],
    summary="List service requests, one page at a time",
    description=(
        "Requests that match every filter, oldest first (by created_at, then id). "
        "Each page has at most `limit` requests. If `next_cursor` is not null, send it "
        "as `cursor`, with the same filters, to get the next page."
    ),
)
def list_service_requests(
    query: Annotated[ListQuery, Query()], con: Database
) -> ServiceRequestPage:
    check_range(query.created_from, query.created_to)
    filters = {
        "borough": query.borough,
        "complaint_type": query.complaint_type,
        "status": query.status,
        "created_from": iso(query.created_from),
        "created_to": iso(query.created_to),
    }
    after = decode_cursor(query.cursor, filters) if query.cursor else None
    rows = queries.list_requests(con, filters, after, query.limit)
    page = rows[: query.limit]
    next_cursor = None
    if len(rows) > query.limit:
        last = page[-1]
        next_cursor = encode_cursor(last["created_at"], last["id"], filters)
    return ServiceRequestPage(items=[dict(row) for row in page], next_cursor=next_cursor)

The limit is not part of the fingerprint. So a client may read a first page of 5 rows, and the rest in pages of 200.

Keyset pages: the cursor is a bookmark after the last row, and the next page starts there.

Text description of the diagram

The dashboard asks for the first page of Bronx requests, 50 at a time. The API asks SQLite for 51 rows, in the order of created_at and then id. 51 rows come back, so a next page exists. The API sends the first 50 rows, and a cursor that holds the created_at and id of row 50. The dashboard sends the same request again, with the cursor. The API decodes it, and checks that the filters are the same. Then it asks SQLite for the rows after that pair, again 51 at most. SQLite finds the place in the index at once. On the last page, fewer than 51 rows come back, and next_cursor is null.

Follow the cursor by hand. Ask for the first two Staten Island requests again, and copy the next_cursor from the answer into the second request. Then send the same cursor with another borough:

Second terminal (on Windows: curl.exe)

curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?borough=staten_island&created_from=2025-03-03T07:00:00&limit=2"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?borough=staten_island&created_from=2025-03-03T07:00:00&limit=2&cursor=eyJjIjoiMjAyNS0wMy0wM1QwNzowMDowMCIsImkiOjY0MjUwNDcxLCJmIjoiNTIxODUyYWEifQ"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/service-requests?borough=queens&limit=2&cursor=eyJjIjoiMjAyNS0wMy0wM1QwNzowMDowMCIsImkiOjY0MjUwNDcxLCJmIjoiNTIxODUyYWEifQ"

Output (each request shortened to its id and created_at)

{"items":[{"id":64250466,"created_at":"2025-03-03T07:00:00",...},{"id":64250471,"created_at":"2025-03-03T07:00:00",...}],"next_cursor":"eyJjIjoiMjAyNS0wMy0wM1QwNzowMDowMCIsImkiOjY0MjUwNDcxLCJmIjoiNTIxODUyYWEifQ"}
{"items":[{"id":64250476,"created_at":"2025-03-03T07:00:00",...},{"id":64251586,"created_at":"2025-03-03T07:00:00",...}],"next_cursor":"eyJjIjoiMjAyNS0wMy0wM1QwNzowMDowMCIsImkiOjY0MjUxNTg2LCJmIjoiNTIxODUyYWEifQ"}
{"error":{"code":"invalid_cursor","message":"The cursor belongs to other filters. Start again without a cursor.","request_id":"5526ce765f49","fields":["cursor"]}}

What this tells us. Page 1 ends with request 64250471 at 07:00:00, inside the batch of 34. Page 2 continues with 64250476, the next ID at the same time: no request of the batch is skipped or repeated. Decoded, the cursor is {"c":"2025-03-03T07:00:00","i":64250471,"f":"521852aa"}: exactly the bookmark. The same cursor with other filters is refused, with a message that tells the client what to do.

Common mistake

The dashboard shows the same requests again and again, or stops after one page

Symptom: Priya’s dashboard shows the first 50 requests again and again, until it stops. Or it shows only 50 requests, though the filter matches thousands.

Cause: the client does not use next_cursor correctly. In the first case, it reads next_cursor, but sends the next request without it, so the API starts from the beginning each time. In the second case, it stops after the first page, instead of when next_cursor is null.

Fix: the client’s loop has one rule. It sends the cursor from the last answer, with the same filters, until next_cursor is null. In Python, with httpx:

cursor = None
while True:
    params = {"borough": "bronx", "limit": 200}
    if cursor:
        params["cursor"] = cursor
    page = httpx.get(url, params=params, headers=headers).json()
    handle(page["items"])
    cursor = page["next_cursor"]
    if cursor is None:
        break

The test test_paging_gives_every_request_once in Part 7 runs this loop against the API.

Part 6: complaint types and daily numbers

Complaint types, for the menus

The dashboards need a menu of complaint types, with the most common first. That is one GROUP BY. Add it to city_api/queries.py:

city_api/queries.py (add)

def complaint_types(con: sqlite3.Connection) -> list[sqlite3.Row]:
    return con.execute(
        "SELECT complaint_type, COUNT(*) AS count FROM service_requests "
        "GROUP BY complaint_type ORDER BY count DESC, complaint_type"
    ).fetchall()

complaint_type as the second sort key gives a fixed order when two types have the same count, so the menu never changes between two calls.

Daily numbers: what to count

The Monday review asks three questions for each day. How many requests came in? How many of them did the crews handle fast? How long did they usually take? Each question needs an exact definition, or two dashboards count differently again.

• opened: the requests created on that day.
• closed_within_14_days: of those, the ones with a closed_at and days_to_close of 14 or less. Fourteen days is the team’s target. The API uses closed_at, not status, because the city fills in the status inconsistently: 203 requests have a close time but a status such as assigned. The SQL course’s case study uses the same definition, so both courses count the same requests.
• share_closed_within_14_days: the second number divided by the first, from 0 to 1.
• median_days_to_close: the middle value of days_to_close of the requests with a closed_at.

Why the median and not the mean? Look at the eleven requests of 3 March in the test data of Part 7. Ten of them closed within 4 days, and one curb repair took 60.98 days. The mean is 6.79 days, which is longer than ten of the eleven requests took. The median is 1.15 days: half closed faster, half slower. One slow request moves the mean a lot, but the median only a little. For “how long does a request usually take”, the median is the honest answer, and the share closed within 14 days shows the slow ones.

SQLite has no MEDIAN function. So the SQL counts, and Python computes each median from the values, with statistics.median. Add this function to city_api/queries.py:

city_api/queries.py (add)

def daily_stats(con: sqlite3.Connection, filters: dict, start: date, end: date) -> list[dict]:
    """Opened, closed within 14 days and the median days to close, for each day."""
    condition, params = where(
        {**filters, "created_from": start.isoformat(), "created_to": end.isoformat()}
    )
    counts = con.execute(
        "SELECT substr(created_at, 1, 10) AS day, COUNT(*) AS opened, "
        "SUM(CASE WHEN closed_at IS NOT NULL AND days_to_close <= 14 THEN 1 ELSE 0 END) AS within_14 "
        f"FROM service_requests WHERE {condition} GROUP BY day",
        params,
    ).fetchall()
    # SQLite has no MEDIAN function, so Python computes it from the values.
    durations: dict[str, list[float]] = {}
    for day, days in con.execute(
        "SELECT substr(created_at, 1, 10), days_to_close FROM service_requests "
        f"WHERE {condition} AND closed_at IS NOT NULL",
        params,
    ):
        durations.setdefault(day, []).append(days)

    by_day = {row["day"]: row for row in counts}
    result = []
    for offset in range((end - start).days):
        day = (start + timedelta(days=offset)).isoformat()
        row = by_day.get(day)
        opened = row["opened"] if row else 0
        within = row["within_14"] if row else 0
        values = durations.get(day)
        result.append(
            {
                "date": day,
                "opened": opened,
                "closed_within_14_days": within,
                "share_closed_within_14_days": round(within / opened, 3) if opened else None,
                "median_days_to_close": round(statistics.median(values), 2) if values else None,
            }
        )
    return result

Part	Why
substr(created_at, 1, 10)	The first 10 characters of 2025-03-03T07:00:00 are the day, 2025-03-03.
SUM(CASE WHEN ... THEN 1 ELSE 0 END)	Counts the rows where the condition is true.
The loop over every day	A day without requests is in the result, with opened 0. A chart with a missing day draws a straight line over it, and a reader does not see the gap.
None for the share and the median	0 divided by 0 has no answer. null says “no data”, but 0 would say “nothing was closed”, which is false.

The statistics endpoint needs a limit on the range. A request for five years of days, for each borough, is a lot of work for one call, and nobody reads such a chart. The dashboards show at most a quarter, so the limit is 92 days, the longest quarter. A longer range gets its own code, range_too_large, so the client knows what to change. Add the query model to city_api/models.py, after ListQuery:

city_api/models.py (add, after ListQuery)

class StatsQuery(RequestFilters):
    created_from: date = Field(description="The first day, for example 2025-03-01.")
    created_to: date = Field(
        description="The day after the last day (not included). At most 92 days after created_from."
    )

Add the response models after ServiceRequestPage:

city_api/models.py (add, after ServiceRequestPage)

class ComplaintTypeCount(BaseModel):
    complaint_type: str
    count: int = Field(description="Requests of this type in the whole database.")


class ComplaintTypeList(BaseModel):
    items: list[ComplaintTypeCount] = Field(description="The most common type first.")


class DailyStat(BaseModel):
    date: date
    opened: int = Field(description="Requests created on this day.")
    closed_within_14_days: int = Field(
        description="Of those, the ones with a closed_at at most 14 days after created_at."
    )
    share_closed_within_14_days: float | None = Field(
        description="closed_within_14_days / opened, from 0 to 1. null when opened is 0."
    )
    median_days_to_close: float | None = Field(
        description="The middle value of days_to_close of the requests with a closed_at. null when none."
    )


class DailyStats(BaseModel):
    """One row for every day of the range, also days without requests."""

    model_config = ConfigDict(
        json_schema_extra={
            "examples": [
                {
                    "items": [
                        {
                            "date": "2025-03-03",
                            "opened": 752,
                            "closed_within_14_days": 680,
                            "share_closed_within_14_days": 0.904,
                            "median_days_to_close": 1.52,
                        }
                    ]
                }
            ]
        }
    )

    items: list[DailyStat]


class DatasetInfo(BaseModel):
    """Where the data comes from, and how it was changed."""

    source: str
    pulled_on: str = Field(description="The snapshot date: statuses are as of this day.")
    created_from: str
    created_to: str = Field(description="Not included.")
    rows: int
    rules: str

DatasetInfo is for one more small endpoint, /v1/dataset. With it, every dashboard can show “data as of 2026-10-07”, which the managers must see. Its query reads the table meta. Add it to city_api/queries.py:

city_api/queries.py (add)

def dataset_info(con: sqlite3.Connection) -> dict[str, str]:
    return dict(con.execute("SELECT name, value FROM meta").fetchall())

Now the three routes. In city_api/routes.py, the line from city_api.models import ErrorResponse, ListQuery, ServiceRequest, ServiceRequestPage imports four models. Replace that one line with this complete list of eight:

city_api/routes.py (replace the import of the models)

from city_api.models import (
    ComplaintTypeList,
    DailyStats,
    DatasetInfo,
    ErrorResponse,
    ListQuery,
    ServiceRequest,
    ServiceRequestPage,
    StatsQuery,
)

The other imports stay as they are. Then add the constant MAX_STATS_DAYS after the last import, from city_api.security import require_api_key, with an empty line before it:

city_api/routes.py (add after the imports)

MAX_STATS_DAYS = 92

Then add the routes at the end of the file:

city_api/routes.py (add at the end)

@router.get(
    "/complaint-types",
    tags=["service requests"],
    summary="List the complaint types, with the number of requests",
)
def list_complaint_types(con: Database) -> ComplaintTypeList:
    return ComplaintTypeList(items=[dict(row) for row in queries.complaint_types(con)])


@router.get(
    "/stats/daily",
    tags=["statistics"],
    summary="Count the requests opened and closed for each day",
    description=(
        "For each day from `created_from` to the day before `created_to`: the requests "
        "opened, how many of them were closed within 14 days, and the median days to "
        f"close. The range is at most {MAX_STATS_DAYS} days. Days without requests are "
        "included, with `opened` 0."
    ),
)
def daily_stats(query: Annotated[StatsQuery, Query()], con: Database) -> DailyStats:
    check_range(query.created_from, query.created_to)
    if (query.created_to - query.created_from).days > MAX_STATS_DAYS:
        raise ApiError(
            422,
            "range_too_large",
            f"The range is longer than {MAX_STATS_DAYS} days. Ask for a shorter one.",
            ["created_from", "created_to"],
        )
    filters = {"borough": query.borough, "complaint_type": query.complaint_type}
    items = queries.daily_stats(con, filters, query.created_from, query.created_to)
    return DailyStats(items=items)


@router.get(
    "/dataset",
    tags=["statistics"],
    summary="Where the data comes from, and its snapshot date",
)
def dataset(con: Database) -> DatasetInfo:
    return DatasetInfo(**queries.dataset_info(con))

The statistics share RequestFilters with the list, so ?borough=Bronx and unknown parameters fail here in the same way. Try the menu and the numbers:

Second terminal (on Windows: curl.exe)

curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/complaint-types"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/stats/daily?created_from=2025-03-01&created_to=2025-03-08&complaint_type=Sidewalk%20Condition"
curl -s -H "X-API-Key: local-dev-key" "http://127.0.0.1:8000/v1/stats/daily?created_from=2025-01-01&created_to=2025-06-01"

Output (the complaint types shortened to the first four)

{"items":[{"complaint_type":"Street Condition","count":39381},{"complaint_type":"Traffic Signal Condition","count":26576},{"complaint_type":"Street Light Condition","count":14342},{"complaint_type":"Sidewalk Condition","count":12061},...]}
{"items":[{"date":"2025-03-01","opened":37,"closed_within_14_days":30,"share_closed_within_14_days":0.811,"median_days_to_close":9.81},{"date":"2025-03-02","opened":15,"closed_within_14_days":10,"share_closed_within_14_days":0.667,"median_days_to_close":8.62},{"date":"2025-03-03","opened":77,"closed_within_14_days":62,"share_closed_within_14_days":0.805,"median_days_to_close":3.94},{"date":"2025-03-04","opened":98,"closed_within_14_days":78,"share_closed_within_14_days":0.796,"median_days_to_close":7.71},{"date":"2025-03-05","opened":100,"closed_within_14_days":54,"share_closed_within_14_days":0.54,"median_days_to_close":12.87},{"date":"2025-03-06","opened":70,"closed_within_14_days":51,"share_closed_within_14_days":0.729,"median_days_to_close":9.43},{"date":"2025-03-07","opened":65,"closed_within_14_days":55,"share_closed_within_14_days":0.846,"median_days_to_close":10.3}]}
{"error":{"code":"range_too_large","message":"The range is longer than 92 days. Ask for a shorter one.","request_id":"895aff6d300a","fields":["created_from","created_to"]}}

What this tells us. There are 31 complaint types, and four of them make 84% of all requests. Street conditions, such as potholes, are the largest group. The week of sidewalk requests shows why the managers want this endpoint. Sidewalks are slow: the daily median is 4 to 13 days, and only 54% of the requests from 5 March were closed within 14 days. Asked for in the same way for the whole of March, the four large types give a clear picture. 95% to 96% of the street-condition and traffic-signal requests closed within 14 days. Only 78% of the sidewalk requests did, and 73% of the street-light requests. That is the kind of difference that the Monday review is for. The five-month range is refused with a clear code. A 92-day range takes about 55 ms on a laptop.

Part 7: tests

A small database for the tests

The tests must not use the 36 MB download. It is slow to get, a test must not depend on the network, and you cannot check numbers by hand on 109,915 rows. So tests/fixture_rows.csv holds 30 real rows of the database, chosen to contain the hard cases:

• six requests of the 07:00:00 batch on 3 March, all at the same second;
• requests from 1 to 5 March of several types, statuses and boroughs;
• two requests without a borough, and a pending request on 1 March;
• an assigned request with a close time, and a curb repair that took 60.98 days;
• a second pending request on 12 March, a week after the others, so that the days between have no requests.

tests/conftest.py builds the database from these rows with the same schema.sql as the real one, in a temporary folder, once per test run. Each test client gets its settings, as in Configuration and logging: Tests with their own settings. The starter’s conftest writes the schema version as the number 1. Now that db.py exists, use its constant instead. In tests/conftest.py, add from city_api.db import SCHEMA_VERSION to the imports, and change the line with PRAGMA user_version:

tests/conftest.py (change)

    con.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")

The most important test: page through everything

The promise of Part 5 is “no request skipped, none repeated”. A test can check exactly that. It follows next_cursor to the end, collects every ID, and compares them with one big page. It does this with pages of 1, 2, 4, 7, 29, 30 and 31 rows. Some of these sizes divide 30, some do not, and some are larger. Here is the start of tests/test_pagination.py:

tests/test_pagination.py (start)

import base64
import json

import pytest


def walk(client, limit, **filters) -> tuple[list[int], int]:
    """Follow next_cursor to the end. Return every id and the number of pages."""
    ids, pages, cursor = [], 0, None
    while True:
        params = {"limit": limit, **filters}
        if cursor:
            params["cursor"] = cursor
        response = client.get("/v1/service-requests", params=params)
        assert response.status_code == 200, response.text
        body = response.json()
        assert len(body["items"]) <= limit
        ids += [item["id"] for item in body["items"]]
        pages += 1
        cursor = body["next_cursor"]
        if cursor is None:
            return ids, pages


def one_page(client, **filters) -> list[int]:
    response = client.get("/v1/service-requests", params={"limit": 200, **filters})
    return [item["id"] for item in response.json()["items"]]


@pytest.mark.parametrize("limit", [1, 2, 4, 7, 29, 30, 31])
def test_paging_gives_every_request_once(client, limit):
    ids, pages = walk(client, limit)
    assert len(ids) == len(set(ids)) == 30  # no duplicates, no gaps
    assert ids == one_page(client)  # in the same order
    assert pages == -(-30 // limit)  # 30 / limit, rounded up

len(ids) == len(set(ids)) == 30 checks both promises at once. A set keeps each ID once, so equal lengths mean no duplicates, and 30 means no gaps. -(-30 // limit) is 30 divided by the limit, rounded up. For limit=4, 7 pages of 4 rows and 1 page of 2 make 8 pages. So the test also proves that no empty last page is sent. The rest of the file has five more tests. They split the 07:00:00 batch, use filters, change the limit, and send broken cursors and cursors from other filters.

The rules, as tests

Three more tests guard the rules of Parts 2 to 4. They are in tests/test_service_requests.py:

tests/test_service_requests.py (three of its tests)

def test_sql_in_a_filter_is_only_text(client):
    attack = "x' OR '1'='1"
    response = client.get("/v1/service-requests", params={"complaint_type": attack})
    assert response.status_code == 200
    assert response.json()["items"] == []


def test_the_connection_cannot_write(fixture_db):
    con = db.connect(fixture_db)
    with pytest.raises(sqlite3.OperationalError, match="readonly"):
        con.execute("DELETE FROM service_requests")
    con.close()


@pytest.mark.parametrize(
    ("params", "field"),
    [
        ({"borough": "Queens"}, "borough"),
        ({"borough": "unspecified"}, "borough"),
        ({"status": "done"}, "status"),
        ({"limit": "0"}, "limit"),
        ({"limit": "201"}, "limit"),
        ({"limit": "ten"}, "limit"),
        ({"created_from": "March 3"}, "created_from"),
        ({"created_from": "2025-03-03T07:00:00Z"}, "created_from"),
        ({"complaint_type": "x" * 61}, "complaint_type"),
        ({"boro": "queens"}, "boro"),
    ],
)
def test_invalid_parameters_name_the_field(client, params, field):
    response = client.get("/v1/service-requests", params=params)
    assert response.status_code == 422
    assert response.json()["error"]["code"] == "invalid_request"
    assert response.json()["error"]["fields"] == [field]

Numbers checked by hand

A statistics endpoint can be wrong and still look right. So the test of /v1/stats/daily compares with numbers that you can check on paper, from the 30 rows. Take 4 March: four requests, all closed within 14 days, after 6.02, 1.02, 7.71 and 9.76 days. Sorted, the two middle values are 6.02 and 7.71, and their mean is 6.865. So the median is 6.87 days, and the share is 4 / 4 = 1.0:

tests/test_stats.py (one row of test_daily_numbers)

        {"date": "2025-03-04", "opened": 4, "closed_within_14_days": 4,
         "share_closed_within_14_days": 1.0, "median_days_to_close": 6.87},

Run them

Copy four files from the finished project’s tests folder into yours: test_health.py, test_pagination.py, test_service_requests.py and test_stats.py. The new test_health.py adds two tests for /ready: one without a database file, and one with an old schema version. Read each file, then run everything:

python -m pytest

Output (end)

tests/test_health.py .....                                               [  7%]
tests/test_pagination.py ...................                             [ 37%]
tests/test_service_requests.py .................................         [ 89%]
tests/test_stats.py .......                                              [100%]

============================== 64 passed in 0.42s ==============================

What this tells us. 64 tests run in under a second, with no server and no download. They cover every endpoint, every error code except 500, the paging promise with seven page sizes, and the two security rules. When you change the SQL later, a mistake fails a test instead of a dashboard.

Part 8: documentation, the contract and Postman

The reviewed contract

The dashboards depend on the API’s names, types and error codes. As in Keep the documentation true, the project keeps a reviewed copy of the OpenAPI document, and a test compares the API with it. The starter has city_api/export_contract.py. Write the contract:

python -m city_api.export_contract

Output

WARNING city_api: API_KEY is not set: /v1 accepts requests without a key
WARNING city_api: API_KEY is not set: /v1 accepts requests without a key
Wrote contract/openapi.json

The warnings are expected: the script makes the app without your terminal’s settings, only to read its OpenAPI document. Read the file before you trust it. Check, for example, that /v1/service-requests lists every query parameter with its description, that borough has an enum with the five codes, and that limit has minimum: 1 and maximum: 200. Open http://127.0.0.1:8000/docs to see the same document in Swagger UI.

Then make tests/test_contract.py:

tests/test_contract.py

import json
from pathlib import Path

CONTRACT = Path(__file__).parent.parent / "contract" / "openapi.json"


def test_openapi_matches_the_reviewed_contract(client):
    current = json.dumps(client.get("/openapi.json").json(), indent=2) + "\n"
    reviewed = CONTRACT.read_text(encoding="utf-8")
    assert current == reviewed, (
        "The API contract changed. If the change is intended, run "
        "python -m city_api.export_contract and review the difference."
    )


def test_documented_request_is_the_real_one(client):
    schemas = client.get("/openapi.json").json()["components"]["schemas"]
    example = schemas["ServiceRequest"]["examples"][0]
    response = client.get(f"/v1/service-requests/{example['id']}")
    assert response.json() == example


def test_documented_examples_have_the_real_shape(client):
    # The examples show numbers of the full database; the tests have 30 rows.
    # So the test compares the fields, not the values.
    schemas = client.get("/openapi.json").json()["components"]["schemas"]
    page = client.get("/v1/service-requests", params={"limit": 1}).json()
    assert page.keys() == schemas["ServiceRequestPage"]["examples"][0].keys()
    day = client.get(
        "/v1/stats/daily", params={"created_from": "2025-03-03", "created_to": "2025-03-04"}
    ).json()["items"][0]
    assert day.keys() == schemas["DailyStats"]["examples"][0]["items"][0].keys()

The example request, 64246800, is one of the 30 test rows, so a test can check it exactly. The daily example shows the real numbers of 3 March, which only the full database has. So that test checks the fields, not the values. Run python -m pytest again: 67 passed.

A README for the dashboard team

A client should not need the code. The finished project’s README.md says what the API is for, and how to set it up. It describes each endpoint in one line, lists the error codes, and shows how to run the checks. Its last section names the data’s source, snapshot date and licence. Copy it, and change it where your project is different.

The Postman collection

The finished project has postman/city-requests-api.postman_collection.json, with 14 requests in three folders. It also has the Local and Deployed Postman environments without keys, as in the course. Most requests check a status and a body, as in Module 5. Two of them do something new: they test the paging across two requests. The first request saves values from its answer in collection variables, and the second request uses them:

Tests of 'List the first page' (the end)

pm.collectionVariables.set("nextCursor", body.next_cursor);
pm.collectionVariables.set("firstPageIds", JSON.stringify(body.items.map((r) => r.id)));
pm.collectionVariables.set("lastCreatedAt", body.items[body.items.length - 1].created_at);

Tests of 'Get the next page' (its URL ends with &cursor={{nextCursor}})

const items = pm.response.json().items;
const firstPage = JSON.parse(pm.collectionVariables.get("firstPageIds"));
pm.test("no request of page 1 again", () => items.forEach((r) => pm.expect(firstPage).to.not.include(r.id)));
pm.test("continues after page 1", () => pm.expect(items[0].created_at >= pm.collectionVariables.get("lastCreatedAt")).to.be.true);

Copy the folder postman from the finished project. With the API running with API_KEY=local-dev-key, run the collection from the second terminal:

postman collection run postman/city-requests-api.postman_collection.json -e postman/local.postman_environment.json --env-var "apiKey=local-dev-key"

Output (start and end)

No authorization data found. Please use the `postman login` command.
Refer:https://learning.postman.com/docs/postman-cli/postman-cli-options/#postman-login

Running your collection...

City Service Requests API

Folder service requests
Sub-folder List the first page

  GET http://127.0.0.1:8000/v1/service-requests?borough=brooklyn&complaint_type=Street%20Condition&created_from=2025-03-01&created_to=2025-04-01&limit=3 [200 OK, 1.38 KB, 17 ms]
  Pass  response is JSON
  Pass  status is 200
  Pass  has 3 requests
  Pass  every request matches the filters
  Pass  has a next_cursor
  Pass  answers in under 500 ms
...
|                requests |                 14 |                 0 |
|              assertions |                 57 |                 0 |

The first lines only say that you are not signed in to Postman, which a local file does not need. What this tells us. 14 requests and 57 checks pass against the real database, through a real server. pytest proved the code with 30 rows. The collection proves the running service with the full data, and it is the smoke check for a deployment. To use it in the Postman app, import the three files in postman/ (Import).

Part 9: deploy it (steps not run by us)

Not run for this case study

We did not deploy this API. The steps below follow the course’s deployment to FastAPI Cloud in Publish safely and Verify and clean up. One step is new and untested: uploading the 36 MB database with the code. FastAPI Cloud is in beta, so commands and limits can change. If you deploy, treat the service as disposable, and delete it when you have your evidence.

What is different from the ticket API

The ticket API had only code. This API also needs its data file on the server. Git and the deploy command leave out the files that .gitignore lists. .gitignore lists data/*.db, because a 36 MB file does not belong in a Git repository. FastAPI Cloud also reads a file .fastapicloudignore, which can bring an ignored file back with !. The starter does not have it. Create .fastapicloudignore in the project folder (the folder with pyproject.toml), with this content. The finished project has the same file:

.fastapicloudignore

# FastAPI Cloud reads this file together with .gitignore.
# Git ignores the database because it is large, but the server needs it:
!data/service_requests_2025h1.db
# The server does not need these:
tests/
postman/
scripts/

The upload then has the code and the database, but not the tests and tools. Because the file is opened with immutable=1, new data means a new deployment: the platform starts a new copy of the app with the new file. That is the safe way to publish new data. In our local test, the running API used less than 80 MB of memory, well under the 500 MiB of a free instance.

The server also builds your project as a Python package. This upload has three top-level folders, city_api/, data/ and contract/, and the build tool cannot guess which one is the package. The live deployment of the house-price case study failed for exactly this reason. So pyproject.toml names the package:

pyproject.toml

[tool.setuptools]
# The project has more than one top-level folder: city_api/ (the code),
# data/ and contract/. Name the package, so the build does not guess.
packages = ["city_api"]

Check it on your computer before you deploy: python -m pip wheel . --no-deps -w build must end with Successfully built city-requests-api. Without those three lines, it fails with Multiple top-level packages discovered in a flat-layout. Then delete the two folders that the check made, build and city_requests_api.egg-info.

Run it as the server will

First, the local deployment simulation from Hosting bridge: production mode, documentation off, and the collection.

macOS

API_KEY=local-dev-key SHOW_DOCS=false fastapi run

Windows

$env:API_KEY = "local-dev-key"
$env:SHOW_DOCS = "false"
fastapi run

Linux

API_KEY=local-dev-key SHOW_DOCS=false fastapi run

Run the collection from a second terminal, as in Part 8. All 14 requests pass. Stop the server.

Deploy, verify, clean up

1. Sign in. Then create the app and link the folder to it, before the first deployment:

   fastapi login
   fastapi cloud apps create --name city-requests-api --link

2. Make a new random key, and store it as a secret. Turn the documentation off:

   macOS

   KEY=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
   printf %s "$KEY" | fastapi cloud env set API_KEY --secret --value-stdin --no-redeploy
   fastapi cloud env set SHOW_DOCS false --no-redeploy
   echo "$KEY"

   Windows

   $key = python -c "import secrets; print(secrets.token_urlsafe(32))"
   $key | fastapi cloud env set API_KEY --secret --value-stdin --no-redeploy
   fastapi cloud env set SHOW_DOCS false --no-redeploy
   $key

   Linux

   KEY=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
   printf %s "$KEY" | fastapi cloud env set API_KEY --secret --value-stdin --no-redeploy
   fastapi cloud env set SHOW_DOCS false --no-redeploy
   echo "$KEY"

   Copy the key that the last command shows, and close the terminal.

3. Open a new terminal, and activate the environment. Then deploy, and write down the address that the command prints:

   fastapi deploy

4. Check the three basic answers. Replace the address with yours. On Windows, type curl.exe:

   curl -i https://city-requests-api.fastapicloud.dev/health
   curl -i https://city-requests-api.fastapicloud.dev/ready
   curl -i https://city-requests-api.fastapicloud.dev/v1/complaint-types

   You expect 200, 200 and 401. If /ready gives 503 database_unavailable, the database did not reach the server. Check .fastapicloudignore, and read the logs (step 6).

5. Set your address as baseUrl in postman/deployed.postman_environment.json, without a / at the end. Then run the smoke check, with the key typed as input:

   macOS

   read -s CITY_API_KEY
   postman collection run postman/city-requests-api.postman_collection.json -e postman/deployed.postman_environment.json --env-var "apiKey=$CITY_API_KEY"

   Windows

   $CITY_API_KEY = Read-Host "API key"
   postman collection run postman/city-requests-api.postman_collection.json -e postman/deployed.postman_environment.json --env-var "apiKey=$CITY_API_KEY"

   Linux

   read -s CITY_API_KEY
   postman collection run postman/city-requests-api.postman_collection.json -e postman/deployed.postman_environment.json --env-var "apiKey=$CITY_API_KEY"

   The first call after a pause can be slow, because the platform stopped the app (scale to zero). If only “answers in under 500 ms” fails, run the collection again.

6. Read the logs, and find a request by its ID: fastapi cloud logs --no-follow --since 10m.

7. Clean up. Delete the app in the FastAPI Cloud dashboard, and run fastapi cloud unlink. Delete the key from Postman, and check that /health no longer answers.

If a dashboard in a browser must call the API directly, set ALLOWED_ORIGINS to the dashboard’s address, as in Browser access and CORS. But every user of that page can see a key in it, and CORS is not authentication. For a real team, the dashboard’s own server calls the API, and keeps the key.

Responsible use and limits

An API makes data easy to use, and that includes easy to misread. The README and /v1/dataset must carry these limits to every dashboard:

• A request is not a problem. Several people report the same pothole, and the city closes many such requests as duplicates, often at once. 4,890 requests have the same time for their creation and their closing. Count requests, and call them requests.
• Who reports. 311 shows where people report, not only where the streets are bad. A neighbourhood where fewer people use 311 can look better than it is. Do not rank neighbourhoods by request counts as if they measured road quality.
• Closed is not repaired. A request can be closed because an inspector found no problem, or because another agency is responsible. The cleaning removed the text that says which. “Closed within 14 days” measures the handling of requests, not repairs.
• A snapshot. Statuses are as of 2026-10-07, and the data ends on 30 June 2025. A dashboard must show the snapshot date, from /v1/dataset.
• Times without a time zone. All times are New York local time. A client that treats them as UTC is wrong by four or five hours.
• Location. The coordinates point to a street or an intersection, not to a person, and the data has no name or phone number of a reporter. Still, do not join it with other data to find out who reported what.
• What this API does not have. It has one key for the whole team, and no rate limit, so one client can send as many requests as it likes. Per-client keys and rate limits are the next steps for a real service. User login is taught in a later course.

Make it a portfolio piece

This project shows skills that employers look for. It has a database behind an API, safe SQL, and pagination that stays correct when data changes. It also has validated inputs, tests with a fixture, a reviewed contract and a deployment plan.

What to put in the repository:

city-requests-api/
  README.md               what it is, the endpoints, the error codes, how to run and check it
  DATASET.md              the data's source, version, licence, cleaning rules and limits
  LICENSE                 MIT, for the code
  pyproject.toml          the pinned versions (requirements-lock.txt for a local set-up)
  city_api/               the API: db.py, queries.py, pagination.py, routes.py, models.py
  scripts/                get_db.py, build_db.py, and the two OFFSET scripts
  tests/                  the 67 tests, with fixture_rows.csv
  contract/openapi.json   the reviewed contract
  postman/                the collection, and its two environments without keys
  data/                   empty in Git: get_db.py downloads the database

To publish it:

1. Make a new Git repository from your project folder, and push it to GitHub. .gitignore keeps out the virtual environment and the database. Anyone can get the database with python scripts/get_db.py.
2. Keep LICENSE (MIT, for the code) and DATASET.md (the data’s source, licence and changes).
3. Make sure that python -m pytest passes from a fresh clone, and say so in the README.
4. Write the README for a stranger. A README template:

README.md (outline)

# city-requests-api

## What it is
A read-only API over 109,915 New York 311 street-repair requests (Jan to Jun 2025),
for dashboards: filtered lists in pages, single requests, complaint types, daily numbers.

## Design choices
- Read-only SQLite (mode=ro, immutable=1); new data means a new deployment.
- Parameterised SQL only; every query in city_api/queries.py.
- Keyset pagination on (created_at, id) with an opaque cursor tied to its filters.
- Strict query parameters: fixed lists, no unknown names, no time zones.

## Run it / Check it
(set-up, pytest, the Postman CLI)

## Data and limits
(source, snapshot date, cleaning rules, what the numbers do not mean)

How to describe it in an interview, in two sentences: “I built a read-only FastAPI service over 110,000 real New York street-repair requests in SQLite, for analysts’ dashboards, with parameterised SQL, validated filters and keyset pagination behind an opaque cursor. A test pages through the data with seven page sizes and proves that it never skips or repeats a row, and a reviewed contract and a Postman collection keep the dashboards’ promises.”

Go further: each of these is a good next step, with one main change.

• Newest first. Add order=desc. The keyset condition becomes <, and the cursor must remember the order.
• A daily refresh. Build tomorrow’s file with build_db.py, check it with the tests, and deploy it. Write down how a dashboard that is paging during the switch behaves.
• Statistics by borough. GET /v1/stats/by-borough for a range, with the same definitions as the daily numbers.
• Caching. The data does not change between deployments, so responses can have an ETag or Cache-Control header. Measure how much it saves.

Knowledge check

Check your understanding

A colleague says: 'Our database file never changes while the API runs, so OFFSET pages are fine.' When is this true, and what changes it?

Show the answer

It is true for this exact setup. With one fixed file, OFFSET returns the same rows for the same page, so nothing is skipped or repeated. Only the cost of deep pages remains, and on 110,000 rows that is about 1.5 ms. It stops being true as soon as the data can change between two requests of one client. Examples are a daily refresh during the paging, or a database that other programs write to. Then a row that leaves or enters the list in front of the current page moves every later row, and OFFSET skips or repeats one. A keyset on (created_at, id) does not depend on positions, so it stays correct. Its cost is also flat, so it is the safer default.

Check your understanding

Why does the keyset use (created_at, id) and not created_at alone? Use a number from this data in your answer.

Show the answer

created_at is not unique: 34 requests have exactly 2025-03-03T07:00:00, and 11,489 requests share their time with another one. If a page ends inside such a group, created_at > last skips the rest of the group, and created_at >= last repeats the part that was already sent. The unique id makes the pair unique, so “after the last row” is one exact place in the order. The test test_a_page_can_end_inside_a_group_of_equal_times checks this case.

Check your understanding

The list endpoint builds its WHERE clause with an f-string. Why is this not SQL injection, and which change to the code would make it injection?

Show the answer

The f-string only joins texts from our own code: the conditions in CONDITIONS, and COLUMNS. The values from the request go to SQLite as parameters, the ? marks, so SQLite never reads them as SQL. It would become injection if any part of the SQL text came from the request. One example is a value put into the condition text instead of the parameters. Another is a column name chosen by the client, such as ?sort=created_at, written into ORDER BY. A sort option must be checked against a fixed list in the code, and only the chosen constant goes into the SQL.

Recap

• A read-only connection (mode=ro) makes a write impossible, not only unlikely. immutable=1 removes the locking, and new data means a restart.
• Values from a request reach SQL only as ? parameters. Column names and SQL words come only from the code.
• A Pydantic model for the query string, with fixed lists and extra="forbid", turns silent wrong answers into clear 422 errors.
• OFFSET pages skip or repeat rows when the data changes, and get slower with depth. A keyset on a unique pair, such as (created_at, id), with an opaque cursor, does neither.
• Statistics need exact definitions: what is counted, the median for durations, null instead of 0 when there is no data, and a limit on the range.
• A small fixture of real rows lets tests check numbers by hand and page through everything in a second.

You built a data API that a team can trust. The final assignment asks you to change the course’s ticket API as a team would, from contract to deployment.

Next: Final assignment