Intermediate Examples

"""Example 29: Validation -- Required Field."""
 
from fastapi import FastAPI  # => the web framework whose validation layer this example exercises
from pydantic import BaseModel  # => Pydantic models are FastAPI's validation vocabulary
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
class TaskCreate(BaseModel):  # => the shape of a valid POST /tasks body
    title: str  # => no default value -- Pydantic treats this as REQUIRED
 
 
@app.post("/tasks", status_code=201)  # => co-08: a handler for creating a task
def create_task(task: TaskCreate) -> dict[str, str]:  # => "task" only exists if validation passed
    # => FastAPI parses+validates the JSON body against TaskCreate BEFORE this
    #    line ever runs -- an invalid body never reaches this function body
    return {"title": task.title}  # => echoes back the validated field

$ curl -s -w "\nHTTP %{http_code}\n" -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{}'
{"detail":[{"type":"missing","loc":["body","title"],"msg":"Field required","input":{}}]}
HTTP 422

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_missing_title_returns_422 PASSED                       [ 50%]
test_app.py::test_valid_title_returns_201 PASSED                         [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.14s =========================

"""Example 30: Validation -- Wrong Type."""
 
from fastapi import FastAPI  # => the web framework whose validation layer this example exercises
from pydantic import BaseModel  # => Pydantic models are FastAPI's validation vocabulary
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
class TaskCreate(BaseModel):  # => the shape of a valid POST /tasks body
    title: str  # => any string is acceptable here
    priority: int  # => MUST be an int -- a non-numeric string fails validation
 
 
@app.post("/tasks", status_code=201)  # => co-08: a handler for creating a task
def create_task(task: TaskCreate) -> dict[str, str | int]:  # => the union return type: str or int values
    # => co-10: FastAPI parses+validates the body against TaskCreate first --
    #    a non-numeric "priority" never reaches this line at all
    return {"title": task.title, "priority": task.priority}  # => both fields, already the right types

$ curl -s -w "\nHTTP %{http_code}\n" -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{"title": "Buy milk", "priority": "urgent"}'
{"detail":[{"type":"int_parsing","loc":["body","priority"],"msg":"Input should be a valid integer, unable to parse string as an integer","input":"urgent"}]}
HTTP 422

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_non_numeric_priority_returns_422 PASSED                [ 50%]
test_app.py::test_numeric_priority_is_coerced_and_accepted PASSED        [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.17s =========================

"""Example 31: Validation -- Field Constraints."""
 
from fastapi import FastAPI  # => the web framework whose validation layer this example exercises
from pydantic import BaseModel, Field  # => Field() attaches constraints beyond a bare type
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
class TaskCreate(BaseModel):  # => the shape of a valid POST /tasks body
    title: str = Field(min_length=1)  # => co-10: rejects an empty string, not just a missing field
    priority: int = Field(gt=0)  # => co-10: rejects zero or negative priorities
 
 
@app.post("/tasks", status_code=201)  # => co-08: a handler for creating a task
def create_task(task: TaskCreate) -> dict[str, str | int]:  # => the union return type: str or int values
    # => both constraints are enforced BEFORE this line runs -- an out-of-range
    #    "priority" or an empty "title" never reaches the handler body
    return {"title": task.title, "priority": task.priority}  # => both fields, already constraint-checked

$ curl -s -w "\nHTTP %{http_code}\n" -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{"title": "", "priority": 0}'
{"detail":[{"type":"string_too_short","loc":["body","title"],"msg":"String should have at least 1 character","input":"","ctx":{"min_length":1}},{"type":"greater_than","loc":["body","priority"],"msg":"Input should be greater than 0","input":0,"ctx":{"gt":0}}]}
HTTP 422

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_empty_title_rejected PASSED                            [ 33%]
test_app.py::test_zero_priority_rejected PASSED                          [ 66%]
test_app.py::test_valid_body_accepted PASSED                             [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 3 passed, 1 warning in 0.15s =========================

"""Example 32: A Custom Error Envelope."""
 
# => FastAPI's DEFAULT 422 body is {"detail": [...]} -- fine for debugging,
#    awkward for a client that expects one stable {"error": {code, message}} shape
from fastapi import FastAPI, Request  # => Request gives the handler access to the raw ASGI request
from fastapi.exceptions import RequestValidationError  # => the exception FastAPI raises internally
from fastapi.responses import JSONResponse  # => the response type this handler builds by hand
from pydantic import BaseModel  # => Pydantic models are FastAPI's validation vocabulary
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
class TaskCreate(BaseModel):  # => the shape of a valid POST /tasks body
    title: str  # => required, no default
 
 
@app.exception_handler(RequestValidationError)  # => overrides FastAPI's DEFAULT 422 handler
async def validation_exception_handler(
    request: Request,  # => unused here, but FastAPI's handler signature always passes it
    exc: RequestValidationError,  # => exc carries every violation FastAPI found
) -> JSONResponse:  # => must return a Response FastAPI can send back to the client
    # => co-11: reshape the default {"detail": [...]} array into a bespoke,
    #    project-specific envelope -- {"error": {"code", "message"}}
    first_error = exc.errors()[0]  # => take the first offending field for a concise message
    field = ".".join(  # => joins the loc TUPLE into one dotted field path string
        str(part)
        for part in first_error["loc"]
        if part != "body"  # => drops the generic "body" segment
    )  # => e.g. "title" (drops the generic "body" location segment)
    return JSONResponse(  # => builds the bespoke envelope BY HAND -- no automatic reshaping exists
        status_code=422,  # => co-03: still 422 -- only the BODY shape changed, not the status
        content={  # => the JSON body FastAPI serializes and sends back to the client
            "error": {  # => nests everything under ONE top-level key, unlike the default array
                "code": "validation_error",  # => a stable, machine-matchable code
                "message": f"{field}: {first_error['msg']}",  # => a human-readable summary
            }  # => closes the inner "error" object
        },  # => closes the "content" dict passed to JSONResponse
    )  # => closes the JSONResponse(...) call itself
 
 
@app.post("/tasks", status_code=201)  # => co-08: a handler for creating a task
def create_task(task: TaskCreate) -> dict[str, str]:  # => "task" only exists if validation passed
    return {"title": task.title}  # => the happy path is untouched by the envelope override

$ curl -s -w "\nHTTP %{http_code}\n" -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{}'
{"error":{"code":"validation_error","message":"title: Field required"}}
HTTP 422

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_missing_title_returns_custom_envelope PASSED           [ 50%]
test_app.py::test_valid_body_is_unaffected PASSED                        [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.16s =========================

"""Example 33: A Domain Exception Handler."""  # => module docstring for this example
 
# => the exception class below knows NOTHING about HTTP -- no status code, no
# => JSON -- that translation happens in exactly one place, the handler below
from fastapi import FastAPI, Request  # => Request gives the handler access to the raw ASGI request
from fastapi.responses import JSONResponse  # => the response type this handler builds by hand
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
class TaskNotFoundError(Exception):  # => a DOMAIN error -- knows nothing about HTTP
    def __init__(self, task_id: int) -> None:  # => constructor
        self.task_id = task_id  # => the id the caller asked for and did not get
 
 
@app.exception_handler(TaskNotFoundError)  # => co-11: maps the domain error to an HTTP shape
async def task_not_found_handler(  # => co-11: registered handler's signature spans three lines
    request: Request,
    exc: TaskNotFoundError,  # => exc carries the id that was missing
) -> JSONResponse:  # => must return a Response FastAPI can send back to the client
    return JSONResponse(  # => builds the 404 envelope by hand, same shape as Example 32's 422
        status_code=404,  # => co-03: not-found status, decided HERE, not in the handler
        content={  # => the JSON body FastAPI serializes and sends back to the client
            "error": {  # => nests everything under ONE top-level key, matching Example 32's shape
                "code": "task_not_found",  # => a stable, machine-matchable code
                "message": f"task {exc.task_id} does not exist",  # => a human-readable summary
            }  # => closes the inner "error" object
        },  # => closes the "content" dict passed to JSONResponse
    )  # => closes the JSONResponse(...) call itself
 
 
_TASKS: dict[int, str] = {1: "Buy milk", 2: "Walk dog"}  # => a tiny in-memory store for this example
 
 
@app.get("/tasks/{task_id}")  # => co-12: a typed path parameter
def get_task(task_id: int) -> dict[str, str]:  # => task_id arrives already parsed as an int
    if task_id not in _TASKS:  # => the handler stays free of any HTTP-status knowledge
        raise TaskNotFoundError(task_id)  # => co-24: raise the DOMAIN error, let it be mapped
    return {"title": _TASKS[task_id]}  # => the found-case return value

$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks/999
{"error":{"code":"task_not_found","message":"task 999 does not exist"}}
HTTP 404
 
$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks/1
{"title":"Buy milk"}
HTTP 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_existing_task_returns_200 PASSED                       [ 50%]
test_app.py::test_missing_task_returns_mapped_404 PASSED                 [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.19s =========================

"""Example 34: A Consistent 404 Envelope Across Methods."""  # => module docstring for this example
 
# => the same domain error, the same handler, and the same shared helper
# => serve BOTH a GET route and a DELETE route below -- consistency by construction
from fastapi import FastAPI, Request  # => Request gives the handler access to the raw ASGI request
from fastapi.responses import JSONResponse  # => the response type this handler builds by hand
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
class TaskNotFoundError(Exception):  # => the same domain error shape as Example 33
    def __init__(self, task_id: int) -> None:  # => constructor
        self.task_id = task_id  # => the id the caller asked for and did not get
 
 
@app.exception_handler(TaskNotFoundError)  # => co-11: ONE handler, reused by every route below
async def task_not_found_handler(  # => co-11: registered handler's signature spans three lines
    request: Request,
    exc: TaskNotFoundError,  # => exc carries the id that was missing
) -> JSONResponse:  # => must return a Response FastAPI can send back to the client
    return JSONResponse(  # => builds the 404 envelope by hand, once, for every caller
        status_code=404,  # => co-03: identical status for every caller of this handler
        content={  # => the JSON body FastAPI serializes and sends back to the client
            "error": {  # => nests everything under ONE top-level key, matching Example 33's shape
                "code": "task_not_found",  # => a stable, machine-matchable code
                "message": f"task {exc.task_id} does not exist",  # => a human-readable summary
            }  # => closes the inner "error" object
        },  # => closes the "content" dict passed to JSONResponse
    )  # => closes the JSONResponse(...) call itself
 
 
_TASKS: dict[int, str] = {1: "Buy milk"}  # => a tiny in-memory store for this example
 
 
def _require_task(task_id: int) -> str:  # => a small shared helper both routes call
    if task_id not in _TASKS:  # => the SAME check, regardless of which route called it
        raise TaskNotFoundError(task_id)  # => the SAME error, regardless of HTTP method
    return _TASKS[task_id]  # => the found-case return value
 
 
@app.get("/tasks/{task_id}")  # => co-12: GET reads
def get_task(task_id: int) -> dict[str, str]:  # => task_id arrives already parsed as an int
    return {"title": _require_task(task_id)}  # => delegates the existence check to the shared helper
 
 
@app.delete("/tasks/{task_id}", status_code=204)  # => co-02: DELETE removes
def delete_task(task_id: int) -> None:  # => 204 means "no body" -- there is nothing to return
    _require_task(task_id)  # => raises the SAME error for a missing id as GET does
    del _TASKS[task_id]  # => only reached once the id is confirmed to exist

$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks/999
{"error":{"code":"task_not_found","message":"task 999 does not exist"}}
HTTP 404
 
$ curl -s -o /dev/null -w "HTTP %{http_code}\n" -X DELETE http://localhost:8000/tasks/999
HTTP 404
 
$ curl -s -o /dev/null -w "HTTP %{http_code}\n" -X DELETE http://localhost:8000/tasks/1
HTTP 204

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_get_missing_task_returns_envelope PASSED               [ 33%]
test_app.py::test_delete_missing_task_returns_same_envelope_shape PASSED [ 66%]
test_app.py::test_delete_existing_task_succeeds PASSED                   [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 3 passed, 1 warning in 0.15s =========================

"""Example 35: A Repository Module Connected to SQLite."""
 
# => this app.py is intentionally THIN -- every database detail lives in
#    repository.py, and this module only ever calls it, never imports sqlite3
from fastapi import FastAPI  # => the web framework whose handler wraps the repository below
 
import repository  # => co-14: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => co-24: schema setup happens ONCE, at import time, not per-request
 
 
@app.get("/tasks")  # => co-08: a handler for listing tasks
def list_tasks() -> list[dict[str, object]]:  # => a JSON array of plain dicts
    rows = repository.list_tasks()  # => co-14: the handler never writes SQL itself
    return [dict(row) for row in rows]  # => sqlite3.Row -> plain dict, JSON-serializable

"""Repository module for Example 35: connects to SQLite and runs a query."""
 
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => co-14: the ONLY database driver this module needs -- ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    # => a NEW connection every call -- sqlite3.Connection objects are not
    #    safe to share across threads, so nothing here caches one
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not row[1]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema and seeds two starter rows
    # => co-14/co-24: schema + seed data live ONLY in the repository module,
    #    never in app.py -- the handler will not even know this function exists
    DB_PATH.unlink(missing_ok=True)  # => delete any stale file first -- every run starts clean
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape once, for the whole example
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => AUTOINCREMENT guarantees ids only ever go up, never get reused
    connection.executemany(  # => runs the SAME statement once per tuple below
        "INSERT INTO tasks (title) VALUES (?)",
        [("Buy milk",), ("Walk dog",)],  # => two seed rows for list_tasks() to return
    )  # => "?" binds each title as DATA, never as SQL syntax
    connection.commit()  # => without this, the inserts never reach the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def list_tasks() -> list[sqlite3.Row]:  # => co-14: the ONLY function that runs a SELECT
    connection = connect()  # => a fresh connection, scoped to just this call
    rows = connection.execute(  # => reads every row, oldest id first
        "SELECT id, title FROM tasks ORDER BY id"  # => no WHERE clause -- returns every row
    ).fetchall()  # => fetchall() -- a plain list of every matching Row
    connection.close()  # => co-24: connection lifetime is scoped to one call, not shared state
    return rows  # => raw sqlite3.Row objects -- the handler converts them to dicts

$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks
[{"id":1,"title":"Buy milk"},{"id":2,"title":"Walk dog"}]
HTTP 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 1 item
 
test_app.py::test_list_tasks_returns_seeded_rows PASSED                  [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 1 passed, 1 warning in 0.15s =========================

"""Example 36: Parameterized Queries Neutralize Injection."""
 
# => this app.py is intentionally THIN -- the "?" placeholder safety property
#    lives entirely in repository.py, and this module only ever calls it
from fastapi import FastAPI  # => the web framework whose handler wraps the repository below
 
import repository  # => co-14: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, deterministic tasks.db for every run
 
 
@app.get("/tasks/search")  # => co-12: a typed query parameter
def search_tasks(title: str) -> list[dict[str, object]]:  # => "title" is really a SEARCH FRAGMENT
    rows = repository.search_by_title(title)  # => co-14: the "?" placeholder does the escaping
    return [dict(row) for row in rows]  # => sqlite3.Row -> plain dict, JSON-serializable

"""Repository module for Example 36: parameterized search."""
 
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => co-14: the ONLY database driver this module needs -- ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not row[1]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema and seeds two starter rows
    DB_PATH.unlink(missing_ok=True)  # => delete any stale file first -- every run starts clean
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape once, for the whole example
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => AUTOINCREMENT guarantees ids only ever go up, never get reused
    connection.executemany(  # => runs the SAME statement once per tuple below
        "INSERT INTO tasks (title) VALUES (?)",  # => one "?" placeholder, bound per call
        [("Buy milk",), ("Walk dog",)],  # => one row contains "milk", the other does not
    )  # => "?" binds each title as DATA, never as SQL syntax
    connection.commit()  # => without this, the inserts never reach the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def search_by_title(fragment: str) -> list[sqlite3.Row]:  # => a substring search over "title"
    # => co-14/co-20: the "?" PLACEHOLDER, not an f-string, is what makes this safe.
    #    SQLite binds `fragment` as DATA, never as part of the SQL grammar itself,
    #    so it is structurally impossible for fragment to inject a second statement
    connection = connect()  # => a fresh connection, scoped to just this call
    rows = connection.execute(  # => LIKE with wildcards built around the bound value
        "SELECT id, title FROM tasks WHERE title LIKE ? ORDER BY id",  # => the one "?" placeholder
        (f"%{fragment}%",),  # => a TUPLE of bound parameters, matching the one "?"
    ).fetchall()  # => zero or more matching rows, never a raised SQL error
    connection.close()  # => co-24: connection lifetime is scoped to one call, not shared state
    return rows  # => raw sqlite3.Row objects -- possibly empty
 
 
def count_tasks() -> int:  # => used only to PROVE the table survived an injection attempt intact
    connection = connect()  # => a fresh connection, scoped to just this call
    (count,) = connection.execute(  # => a single-row, single-column result, unpacked directly
        "SELECT COUNT(*) FROM tasks"  # => no WHERE clause -- counts every row in the table
    ).fetchone()  # => fetchone() -- exactly one row is always returned by COUNT(*)
    connection.close()  # => releases the connection immediately after reading
    return int(count)  # => COUNT(*) already returns an int, but be explicit for the type checker

$ curl -s -w "\nHTTP %{http_code}\n" -G http://localhost:8000/tasks/search --data-urlencode "title=milk"
[{"id":1,"title":"Buy milk"}]
HTTP 200
 
$ curl -s -w "\nHTTP %{http_code}\n" -G http://localhost:8000/tasks/search --data-urlencode "title='; DROP TABLE tasks; --"
[]
HTTP 200
 
$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks
{"detail":"Not Found"}
HTTP 404

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_normal_search_matches_a_row PASSED                     [ 50%]
test_app.py::test_injection_attempt_matches_nothing_and_table_survives PASSED [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.19s =========================

"""Example 37: CRUD -- Create."""
 
# => the "C" in CRUD -- one POST endpoint, backed by repository.create_task(),
#    the start of the growing task-management service this topic builds
from fastapi import FastAPI  # => the web framework whose handler wraps the repository below
from pydantic import BaseModel  # => Pydantic models are FastAPI's validation vocabulary
 
import repository  # => co-14: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, empty tasks.db for every run
 
 
class TaskCreate(BaseModel):  # => the shape of a valid POST /tasks body
    title: str  # => co-10: validated before the handler ever sees it
 
 
@app.post("/tasks", status_code=201)  # => co-03: 201 signals a new resource was created
def create_task(task: TaskCreate) -> dict[str, object]:  # => "task" only exists if validation passed
    new_id = repository.create_task(task.title)  # => co-14: the handler delegates the INSERT
    return {"id": new_id, "title": task.title}  # => echoes the id the DB actually assigned

"""Repository module for Example 37: insert a new row."""
 
# => the "C" in CRUD -- this file's job is one function, create_task(), and
#    every INSERT detail for the whole example lives right here, nowhere else
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => co-14: the ONLY database driver this module needs -- ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not row[1]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema, starting EMPTY
    DB_PATH.unlink(missing_ok=True)  # => delete any stale file first -- every run starts clean
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape -- no seed rows this time
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => starts EMPTY -- this example is entirely about the create path
    connection.commit()  # => without this, the table definition never reaches the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def create_task(title: str) -> int:  # => co-14/co-08: the ONLY function that runs an INSERT
    connection = connect()  # => a fresh connection, scoped to just this call
    cursor = connection.execute(  # => "?" binds title as DATA, the safety property from Example 36
        "INSERT INTO tasks (title) VALUES (?)",
        (title,),  # => one placeholder, one bound value
    )  # => cursor.lastrowid below reads the id SQLite just assigned
    connection.commit()  # => without this, the row exists only in this connection's transaction
    new_id = cursor.lastrowid  # => sqlite3 exposes the AUTOINCREMENT id straight off the cursor
    connection.close()  # => co-24: connection lifetime is scoped to one call, not shared state
    assert new_id is not None  # => guaranteed non-None right after a successful INSERT
    return new_id  # => the caller needs this to build the 201 response body
 
 
def get_task(task_id: int) -> sqlite3.Row | None:  # => used only to PROVE persistence
    connection = connect()  # => a fresh connection, scoped to just this call
    row = connection.execute(  # => looks up exactly one row by its primary key
        "SELECT id, title FROM tasks WHERE id = ?",
        (task_id,),  # => one placeholder, one bound value
    ).fetchone()  # => a single Row, or None if task_id does not exist
    connection.close()  # => releases the connection immediately after reading
    return row  # => propagates the None case straight through to the caller

$ curl -s -w "\nHTTP %{http_code}\n" -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{"title": "Buy milk"}'
{"id":1,"title":"Buy milk"}
HTTP 201

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_create_returns_201_with_new_id PASSED                  [ 50%]
test_app.py::test_created_row_actually_persists PASSED                   [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.14s =========================

"""Example 38: CRUD -- Read One."""
 
# => the "R" (singular) in CRUD -- one GET endpoint, backed by
#    repository.get_task(), returning a 404 when the id does not exist
from fastapi import FastAPI, HTTPException  # => HTTPException raises FastAPI's default error shape
 
import repository  # => co-14: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, seeded tasks.db for every run
 
 
@app.get("/tasks/{task_id}")  # => co-12: a typed integer path parameter
def get_task(task_id: int) -> dict[str, object]:  # => task_id arrives already parsed as an int
    row = repository.get_task(task_id)  # => co-14: the handler delegates the SELECT
    if row is None:  # => the ONLY branch this handler makes -- everything else lives in the repo
        raise HTTPException(status_code=404)  # => FastAPI's plain default 404 shape
    return dict(row)  # => sqlite3.Row -> plain dict, JSON-serializable

"""Repository module for Example 38: read a single row by id."""
 
# => the "R" (singular) in CRUD -- this file's job is one function, get_task(),
#    and every SELECT-by-id detail for the whole example lives right here
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => co-14: the ONLY database driver this module needs -- ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not row[1]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema and seeds two starter rows
    DB_PATH.unlink(missing_ok=True)  # => delete any stale file first -- every run starts clean
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape once, for the whole example
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => AUTOINCREMENT guarantees ids only ever go up, never get reused
    connection.executemany(  # => runs the SAME statement once per tuple below
        "INSERT INTO tasks (title) VALUES (?)",
        [("Buy milk",), ("Walk dog",)],  # => two seed rows
    )  # => two seed rows -- id 1 and id 2, in insertion order
    connection.commit()  # => without this, the inserts never reach the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def get_task(task_id: int) -> sqlite3.Row | None:  # => co-14/co-12: one row, by its path parameter
    connection = connect()  # => a fresh connection, scoped to just this call
    row = connection.execute(  # => looks up exactly one row by its primary key
        "SELECT id, title FROM tasks WHERE id = ?",
        (task_id,),  # => one placeholder, one bound value
    ).fetchone()  # => fetchone() -- a single Row, or None when nothing matches
    connection.close()  # => co-24: connection lifetime is scoped to one call, not shared state
    return row  # => the handler decides what a None result means (a 404)

$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks/1
{"id":1,"title":"Buy milk"}
HTTP 200
 
$ curl -s -o /dev/null -w "HTTP %{http_code}\n" http://localhost:8000/tasks/999
HTTP 404

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_read_existing_task PASSED                              [ 50%]
test_app.py::test_read_missing_task_returns_404 PASSED                   [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.14s =========================

"""Example 39: CRUD -- Read List."""
 
# => the "R" (plural) in CRUD -- POST grows the list, GET returns the
#    whole array, so a single curl session can prove the two compose
from fastapi import FastAPI  # => the web framework whose handlers wrap the repository below
from pydantic import BaseModel  # => Pydantic models are FastAPI's validation vocabulary
 
import repository  # => co-14: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, empty tasks.db for every run
 
 
class TaskCreate(BaseModel):  # => the shape of a valid POST /tasks body
    title: str  # => co-10: validated before either handler below sees it
 
 
@app.post("/tasks", status_code=201)  # => reused so curl can grow the list before listing it
def create_task(task: TaskCreate) -> dict[str, object]:  # => "task" only exists if validation passed
    new_id = repository.create_task(task.title)  # => co-14: the handler delegates the INSERT
    return {"id": new_id, "title": task.title}  # => echoes the id the DB actually assigned
 
 
@app.get("/tasks")  # => co-14: returns a JSON ARRAY, one element per row
def list_tasks() -> list[dict[str, object]]:  # => the array shape a client can iterate directly
    rows = repository.list_tasks()  # => co-14: the handler never writes SQL itself
    return [dict(row) for row in rows]  # => sqlite3.Row -> plain dict, JSON-serializable

"""Repository module for Example 39: read the whole list."""
 
# => the "R" (plural) in CRUD -- create_task() grows the table, list_tasks()
#    reads every row back, so this one file proves both halves compose
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => co-14: the ONLY database driver this module needs -- ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not row[1]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema, starting EMPTY
    DB_PATH.unlink(missing_ok=True)  # => delete any stale file first -- every run starts clean
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape -- no seed rows this time
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => starts EMPTY -- every row here arrives through create_task() below
    connection.commit()  # => without this, the table definition never reaches the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def create_task(title: str) -> int:  # => co-14: reused so curl/pytest can grow the list first
    connection = connect()  # => a fresh connection, scoped to just this call
    cursor = connection.execute(  # => "?" binds title as DATA, the same property as every example
        "INSERT INTO tasks (title) VALUES (?)",
        (title,),  # => one placeholder, one bound value
    )  # => cursor.lastrowid below reads the id SQLite just assigned
    connection.commit()  # => without this, the row exists only in this connection's transaction
    new_id = cursor.lastrowid  # => sqlite3 exposes the AUTOINCREMENT id straight off the cursor
    connection.close()  # => releases the connection immediately after writing
    assert new_id is not None  # => guaranteed non-None right after a successful INSERT
    return new_id  # => the caller needs this to build the 201 response body
 
 
def list_tasks() -> list[sqlite3.Row]:  # => co-14: every row, in insertion order -- the "R" of CRUD
    connection = connect()  # => a fresh connection, scoped to just this call
    rows = connection.execute(  # => reads every row, oldest id first
        "SELECT id, title FROM tasks ORDER BY id"  # => no WHERE clause -- returns every row
    ).fetchall()  # => fetchall() -- a plain list of every matching Row
    connection.close()  # => co-24: connection lifetime is scoped to one call, not shared state
    return rows  # => raw sqlite3.Row objects -- the handler converts them to dicts

$ curl -s -w "\nHTTP %{http_code}\n" -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{"title": "Buy milk"}'
{"id":1,"title":"Buy milk"}
HTTP 201
 
$ curl -s -w "\nHTTP %{http_code}\n" -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{"title": "Walk dog"}'
{"id":2,"title":"Walk dog"}
HTTP 201
 
$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks
[{"id":1,"title":"Buy milk"},{"id":2,"title":"Walk dog"}]
HTTP 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_list_starts_empty PASSED                               [ 50%]
test_app.py::test_list_grows_as_tasks_are_created PASSED                 [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.17s =========================

"""Example 40: CRUD -- Update."""
 
# => the "U" in CRUD -- one PUT endpoint, backed by repository.update_task(),
#    which reports back whether a row actually changed
from fastapi import FastAPI, HTTPException  # => HTTPException raises FastAPI's default error shape
from pydantic import BaseModel  # => Pydantic models are FastAPI's validation vocabulary
 
import repository  # => co-14: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, seeded tasks.db for every run
 
 
class TaskUpdate(BaseModel):  # => the shape of a valid PUT /tasks/{id} body
    title: str  # => co-10: the replacement value, validated like any other body
 
 
@app.put("/tasks/{task_id}")  # => co-02: PUT replaces the addressed resource
def update_task(task_id: int, task: TaskUpdate) -> dict[str, object]:  # => path param + validated body
    changed = repository.update_task(task_id, task.title)  # => co-14: delegates the UPDATE
    if not changed:  # => the ONLY branch this handler makes -- everything else lives in the repo
        raise HTTPException(status_code=404)  # => FastAPI's plain default 404 shape
    return {"id": task_id, "title": task.title}  # => echoes the id + the value that was just written

"""Repository module for Example 40: update a row in place."""
 
# => the "U" in CRUD -- this file's job is one function, update_task(), which
#    runs an UPDATE and reports back via rowcount whether a row actually changed
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => the ONLY database driver this module needs -- it ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not just row[0]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema and seeds two starter rows
    DB_PATH.unlink(missing_ok=True)  # => start every run from a clean, deterministic file
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape once, for the whole example
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => AUTOINCREMENT guarantees ids only ever go up, never get reused
    connection.executemany(  # => runs the SAME statement once per tuple below
        "INSERT INTO tasks (title) VALUES (?)",  # => one "?" placeholder, bound per tuple
        [("Buy milk",), ("Walk dog",)],  # => two seed rows
    )  # => two seed rows -- id 1 and id 2, both about to be updated
    connection.commit()  # => without this, the inserts never reach the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def update_task(task_id: int, title: str) -> bool:  # => co-14/co-02: UPDATE, not DELETE+INSERT
    connection = connect()  # => a fresh connection, scoped to just this call
    cursor = connection.execute(  # => two "?" placeholders, bound positionally in tuple order
        "UPDATE tasks SET title = ? WHERE id = ?",  # => co-14: parameterized -- never string-formatted
        (title, task_id),  # => title first, then task_id
    )  # => cursor.rowcount below reports how many rows this statement touched
    connection.commit()  # => without this, the change never reaches the file on disk
    changed = cursor.rowcount > 0  # => rowcount is 0 when task_id did not match any row
    connection.close()  # => co-24: connection lifetime is scoped to one call, not shared state
    return changed  # => lets the handler distinguish "updated" from "nothing to update"
 
 
def get_task(task_id: int) -> sqlite3.Row | None:  # => used only to PROVE the update landed on disk
    connection = connect()  # => a fresh connection, scoped to just this call
    row = connection.execute(  # => looks up exactly one row by its primary key
        "SELECT id, title FROM tasks WHERE id = ?",  # => co-14: same parameterized pattern throughout
        (task_id,),  # => one placeholder, one bound value
    ).fetchone()  # => a single Row, or None if task_id does not exist
    connection.close()  # => releases the connection immediately after reading
    return row  # => propagates the None case straight through to the caller

$ curl -s -w "\nHTTP %{http_code}\n" -X PUT http://localhost:8000/tasks/1 -H "Content-Type: application/json" -d '{"title": "Buy oat milk"}'
{"id":1,"title":"Buy oat milk"}
HTTP 200
 
$ curl -s -o /dev/null -w "HTTP %{http_code}\n" -X PUT http://localhost:8000/tasks/999 -H "Content-Type: application/json" -d '{"title": "ghost"}'
HTTP 404

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_update_returns_200_with_new_title PASSED               [ 33%]
test_app.py::test_update_actually_persists PASSED                        [ 66%]
test_app.py::test_update_missing_id_returns_404 PASSED                   [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 3 passed, 1 warning in 0.15s =========================

"""Example 41: CRUD -- Delete."""
 
# => the "D" in CRUD -- one DELETE endpoint, backed by repository.delete_task(),
#    which reports back whether a row actually existed to remove
from fastapi import FastAPI, HTTPException  # => HTTPException raises FastAPI's default error shape
 
import repository  # => co-14: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, seeded tasks.db for every run
 
 
@app.delete("/tasks/{task_id}", status_code=204)  # => co-03: 204 -- success, no body
def delete_task(task_id: int) -> None:  # => task_id arrives already parsed as an int
    removed = repository.delete_task(task_id)  # => co-14: delegates the DELETE
    if not removed:  # => the ONLY branch this handler makes -- everything else lives in the repo
        raise HTTPException(status_code=404)  # => distinguishes "gone" from "never existed"

"""Repository module for Example 41: delete a row."""
 
# => the "D" in CRUD -- this file's job is one function, delete_task(), which
#    runs a DELETE and reports back via rowcount whether a row actually existed
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => the ONLY database driver this module needs -- it ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not just row[0]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema and seeds two starter rows
    DB_PATH.unlink(missing_ok=True)  # => start every run from a clean, deterministic file
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape once, for the whole example
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => AUTOINCREMENT guarantees ids only ever go up, never get reused
    connection.executemany(  # => runs the SAME statement once per tuple below
        "INSERT INTO tasks (title) VALUES (?)",  # => one "?" placeholder, bound per tuple
        [("Buy milk",), ("Walk dog",)],  # => two seed rows
    )  # => two seed rows -- id 1 and id 2, both about to be deleted
    connection.commit()  # => without this, the inserts never reach the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def delete_task(task_id: int) -> bool:  # => co-14: gated on rowcount, same shape as update_task
    connection = connect()  # => a fresh connection, scoped to just this call
    cursor = connection.execute(  # => "?" binds task_id as DATA, never as literal SQL text
        "DELETE FROM tasks WHERE id = ?",  # => co-14: parameterized -- never string-formatted
        (task_id,),  # => one placeholder, one bound value
    )  # => cursor.rowcount below reports how many rows this statement touched
    connection.commit()  # => without this, the deletion never reaches the file on disk
    removed = cursor.rowcount > 0  # => rowcount is 0 when task_id did not match any row
    connection.close()  # => co-24: connection lifetime is scoped to one call, not shared state
    return removed  # => lets the handler distinguish "removed" from "nothing to remove"
 
 
def get_task(task_id: int) -> sqlite3.Row | None:  # => used only to PROVE the row is genuinely gone
    connection = connect()  # => a fresh connection, scoped to just this call
    row = connection.execute(  # => looks up exactly one row by its primary key
        "SELECT id, title FROM tasks WHERE id = ?",  # => co-14: same parameterized pattern throughout
        (task_id,),  # => one placeholder, one bound value
    ).fetchone()  # => a single Row, or None once the row is deleted
    connection.close()  # => releases the connection immediately after reading
    return row  # => propagates the None case straight through to the caller

$ curl -s -o /dev/null -w "HTTP %{http_code}\n" -X DELETE http://localhost:8000/tasks/1
HTTP 204
 
$ curl -s -o /dev/null -w "HTTP %{http_code}\n" -X DELETE http://localhost:8000/tasks/1
HTTP 404

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_delete_returns_204 PASSED                              [ 33%]
test_app.py::test_delete_actually_removes_the_row PASSED                 [ 66%]
test_app.py::test_delete_missing_id_returns_404 PASSED                   [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 3 passed, 1 warning in 0.17s =========================

"""Example 42: CRUD -- Missing Id Returns a 404 Envelope."""  # => module docstring for this example
 
# => combines the DB-backed UPDATE/DELETE from Example 40/41 with the shared
# => envelope pattern from Example 33/34 -- now BOTH sources of truth agree
from fastapi import FastAPI, Request  # => Request gives the handler access to the raw ASGI request
from fastapi.responses import JSONResponse  # => the response type this handler builds by hand
from pydantic import BaseModel  # => Pydantic models are FastAPI's validation vocabulary
 
import repository  # => co-14: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, seeded tasks.db for every run
 
 
class TaskNotFoundError(Exception):  # => the same envelope-shaped domain error as ex-33/ex-34
    def __init__(self, task_id: int) -> None:  # => constructor
        self.task_id = task_id  # => the id the caller asked for and did not get
 
 
@app.exception_handler(TaskNotFoundError)  # => co-11: ONE handler for BOTH routes below
async def task_not_found_handler(  # => co-11: registered handler's signature spans three lines
    request: Request,
    exc: TaskNotFoundError,  # => exc carries the id that was missing
) -> JSONResponse:  # => must return a Response FastAPI can send back to the client
    return JSONResponse(  # => builds the 404 envelope by hand, shared by PUT and DELETE below
        status_code=404,  # => co-03: identical status for every caller of this handler
        content={  # => the JSON body FastAPI serializes and sends back to the client
            "error": {  # => nests everything under ONE top-level key, matching ex-33/ex-34
                "code": "task_not_found",  # => a stable, machine-matchable code
                "message": f"task {exc.task_id} does not exist",  # => a human-readable summary
            }  # => closes the inner "error" object
        },  # => closes the "content" dict passed to JSONResponse
    )  # => closes the JSONResponse(...) call itself
 
 
class TaskUpdate(BaseModel):  # => the shape of a valid PUT /tasks/{id} body
    title: str  # => co-10: validated before the handler ever sees it
 
 
@app.put("/tasks/{task_id}")  # => co-02: PUT against a real, repository-backed store now
def update_task(task_id: int, task: TaskUpdate) -> dict[str, object]:  # => path param + validated body
    changed = repository.update_task(task_id, task.title)  # => co-14: delegates the UPDATE
    if not changed:  # => the ONLY branch this handler makes -- everything else lives in the repo
        raise TaskNotFoundError(task_id)  # => same envelope, now driven by rowcount==0
    return {"id": task_id, "title": task.title}  # => echoes the id + the value that was just written
 
 
@app.delete("/tasks/{task_id}", status_code=204)  # => co-03: 204 -- success, no body
def delete_task(task_id: int) -> None:  # => task_id arrives already parsed as an int
    removed = repository.delete_task(task_id)  # => co-14: delegates the DELETE
    if not removed:  # => the ONLY branch this handler makes -- everything else lives in the repo
        raise TaskNotFoundError(task_id)  # => the SAME error class as the PUT branch above

"""Repository module for Example 42: update/delete against a missing id."""  # => module docstring
 
# => the SAME update_task()/delete_task() shape as ex-40/ex-41, seeded with
# => only ONE row this time so most ids the app queries are genuinely missing
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => the ONLY database driver this module needs -- it ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not just row[0]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema and seeds exactly ONE starter row
    DB_PATH.unlink(missing_ok=True)  # => start every run from a clean, deterministic file
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape once, for the whole example
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => AUTOINCREMENT guarantees ids only ever go up, never get reused
    connection.execute(  # => runs the INSERT exactly once, unlike executemany() elsewhere
        "INSERT INTO tasks (title) VALUES (?)",
        ("Buy milk",),  # => one seed row
    )  # => one seed row, id 1 -- every id this example queries besides 1 is missing
    connection.commit()  # => without this, the insert never reaches the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def update_task(task_id: int, title: str) -> bool:  # => co-14/co-02: identical shape to ex-40
    connection = connect()  # => a fresh connection, scoped to just this call
    cursor = connection.execute(  # => two "?" placeholders, bound positionally in tuple order
        "UPDATE tasks SET title = ? WHERE id = ?",
        (title, task_id),  # => title first, then task_id
    )  # => cursor.rowcount below reports how many rows this statement touched
    connection.commit()  # => without this, the change never reaches the file on disk
    changed = cursor.rowcount > 0  # => rowcount is 0 when task_id did not match any row
    connection.close()  # => co-24: connection lifetime is scoped to one call, not shared state
    return changed  # => drives the app-level TaskNotFoundError below
 
 
def delete_task(task_id: int) -> bool:  # => co-14: identical shape to ex-41's delete_task
    connection = connect()  # => a fresh connection, scoped to just this call
    cursor = connection.execute(  # => "?" binds task_id as DATA, never as literal SQL text
        "DELETE FROM tasks WHERE id = ?",
        (task_id,),  # => one placeholder, one bound value
    )  # => cursor.rowcount below reports how many rows this statement touched
    connection.commit()  # => without this, the deletion never reaches the file on disk
    removed = cursor.rowcount > 0  # => rowcount is 0 when task_id did not match any row
    connection.close()  # => releases the connection immediately after writing
    return removed  # => drives the SAME app-level TaskNotFoundError as update_task above

$ curl -s -w "\nHTTP %{http_code}\n" -X PUT http://localhost:8000/tasks/999 -H "Content-Type: application/json" -d '{"title": "ghost"}'
{"error":{"code":"task_not_found","message":"task 999 does not exist"}}
HTTP 404
 
$ curl -s -w "\nHTTP %{http_code}\n" -X DELETE http://localhost:8000/tasks/999
{"error":{"code":"task_not_found","message":"task 999 does not exist"}}
HTTP 404

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_update_missing_id_returns_envelope PASSED              [ 50%]
test_app.py::test_delete_missing_id_returns_same_envelope_shape PASSED   [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.16s =========================

"""Example 43: Apply a schema.sql Migration at Startup."""  # => module docstring for this example
 
# => the schema lives in a SEPARATE schema.sql file, not a Python string --
# => this app.py only ever asks repository.py "did the migration succeed?"
from fastapi import FastAPI  # => the web framework whose health check wraps the repository below
 
import repository  # => co-15: the module that owns the schema.sql migration for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.apply_schema()  # => co-15: runs BEFORE the app accepts a single request
assert repository.table_exists(  # => fails FAST, at import time, not on the first real request
    "tasks"  # => the single table name this migration creates
), "migration failed: tasks table missing"  # => verified before uvicorn finishes booting --
# => a crash here is far cheaper to diagnose than a 500 on the very first real request
 
 
@app.get("/health")  # => co-08: a readiness-style check a load balancer could poll
def health() -> dict[str, bool]:  # => a tiny, machine-checkable readiness signal
    return {"tasks_table_ready": repository.table_exists("tasks")}  # => re-verified per request

"""Repository module for Example 43: apply schema.sql at startup."""  # => module docstring
 
# => co-15: the schema's SOURCE OF TRUTH is the .sql FILE next to this module,
# => not a Python string -- apply_schema() reads it and runs it as one script
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => the ONLY database driver this module needs -- it ships with Python itself
from pathlib import Path  # => builds absolute, OS-independent paths to the db file and schema.sql
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
SCHEMA_PATH = Path(__file__).parent / "schema.sql"  # => co-15: the schema's SOURCE OF TRUTH
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not just row[0]
    return connection  # => the caller owns closing this connection when done
 
 
def apply_schema() -> None:  # => co-15: reads and runs schema.sql, once, at startup
    # => co-15: open the .sql file and execute it as a whole script -- the schema's
    # => source of truth is the FILE, not a Python string embedded in this module
    DB_PATH.unlink(missing_ok=True)  # => start every run from a clean, deterministic file
    connection = connect()  # => a fresh connection, scoped to just this migration call
    schema_sql = SCHEMA_PATH.read_text()  # => reads the ENTIRE schema.sql file as one string
    connection.executescript(schema_sql)  # => runs every statement in that file, in file order
    connection.commit()  # => without this, the CREATE TABLE never reaches the file on disk
    connection.close()  # => releases the connection -- apply_schema() is a one-shot setup call
 
 
def table_exists(table_name: str) -> bool:  # => co-15: proof the migration genuinely ran
    # => queries SQLite's own catalog table -- proof the migration genuinely ran
    connection = connect()  # => a fresh connection, scoped to just this call
    row = connection.execute(  # => SQLite tracks every table it owns in this catalog table
        "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?",  # => co-14: parameterized
        (table_name,),  # => "?" binds table_name as DATA, not as SQL syntax
    ).fetchone()  # => a matching catalog row, or None if the table does not exist
    connection.close()  # => releases the connection immediately after reading
    return row is not None  # => a plain boolean the startup assertion can check directly

-- Example 43: the schema lives in its own .sql file, not inline Python.
-- The same file could be handed to `sqlite3 tasks.db < schema.sql` on a shell.
CREATE TABLE IF NOT EXISTS tasks ( -- => co-15: begins the single-table DDL statement
  id INTEGER PRIMARY KEY AUTOINCREMENT, -- => co-15: ids only ever go up, never get reused
  title TEXT NOT NULL -- => co-15: the one column this migration creates
);

$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/health
{"tasks_table_ready":true}
HTTP 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_table_exists_after_migration PASSED                    [ 50%]
test_app.py::test_health_endpoint_confirms_readiness PASSED              [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.15s =========================

"""Example 44: An Additive ALTER TABLE + Backfill Migration."""
 
# => two calls at import time: FIRST seed the OLD schema and a row, THEN
#    migrate -- proving the migration works against data that predates it
from fastapi import FastAPI  # => the web framework whose handler wraps the repository below
 
import repository  # => co-15: the module that owns the migration for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.create_v1_schema()  # => v1: (id, title) only, seeded with one pre-migration row
repository.migrate_add_priority_column()  # => co-15: additive ALTER TABLE + backfill UPDATE
 
 
@app.get("/tasks/{task_id}")  # => co-12: a typed integer path parameter
def get_task(task_id: int) -> dict[str, object]:  # => task_id arrives already parsed as an int
    row = repository.get_task(task_id)  # => co-14: delegates the SELECT
    assert row is not None  # => id 1 always exists in this example's fixed seed data
    return dict(row)  # => includes the BACKFILLED "priority" column

"""Repository + migration module for Example 44."""
 
# => co-15: an ADDITIVE migration -- create_v1_schema() seeds the OLD shape,
#    migrate_add_priority_column() adds a column AND backfills existing rows
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => the ONLY database driver this module needs -- it ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not just row[0]
    return connection  # => the caller owns closing this connection when done
 
 
def create_v1_schema() -> None:  # => co-15: the ORIGINAL schema, before this migration runs
    # => co-15: the ORIGINAL schema, before this example's migration ever runs
    DB_PATH.unlink(missing_ok=True)  # => start every run from a clean, deterministic file
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => v1 DDL -- deliberately narrower than the post-migration shape
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact v1 DDL
    )  # => v1: only (id, title) -- no "priority" column exists yet
    connection.execute(  # => runs BEFORE migrate_add_priority_column() below ever executes
        "INSERT INTO tasks (title) VALUES (?)",
        ("Buy milk",),  # => one pre-migration row
    )  # => a row inserted BEFORE the "priority" column exists at all
    connection.commit()  # => without this, the insert never reaches the file on disk
    connection.close()  # => releases the connection -- create_v1_schema() is a one-shot call
 
 
def migrate_add_priority_column() -> None:  # => co-15: additive ALTER TABLE + backfill UPDATE
    # => co-15: ADDITIVE only -- never drops or renames the existing "title" column
    connection = connect()  # => a fresh connection, scoped to just this migration call
    connection.execute(  # => co-15: ADD COLUMN never touches an existing column or row's data
        "ALTER TABLE tasks ADD COLUMN priority INTEGER"  # => the exact additive DDL
    )  # => new column; every existing row gets NULL here, including the one above
    connection.execute(  # => co-15: the explicit BACKFILL step, run right after the ALTER
        "UPDATE tasks SET priority = 3 WHERE priority IS NULL"  # => only touches the NULL rows
    )  # => the explicit BACKFILL step -- gives every pre-existing row a real value
    connection.commit()  # => without this, neither the ALTER nor the backfill persists to disk
    connection.close()  # => releases the connection -- migrate_add_priority_column() is one-shot
 
 
def column_names() -> set[str]:  # => introspects the live schema, used to PROVE the migration ran
    # => introspects the live schema -- used to PROVE the column was absent, then present
    connection = connect()  # => a fresh connection, scoped to just this call
    info = connection.execute(  # => SQLite's own metadata pragma, no separate driver call needed
        "PRAGMA table_info(tasks)"  # => not parameterized -- table names cannot be bound with "?"
    ).fetchall()  # => one row per column, each with a "name" field
    connection.close()  # => releases the connection immediately after reading
    return {row["name"] for row in info}  # => a set is the right shape for a fast "in" check
 
 
def get_task(task_id: int) -> sqlite3.Row | None:  # => used only to PROVE the backfilled value
    connection = connect()  # => a fresh connection, scoped to just this call
    row = connection.execute(  # => reads the POST-migration shape, including "priority"
        "SELECT id, title, priority FROM tasks WHERE id = ?",
        (task_id,),  # => one bound value
    ).fetchone()  # => includes the BACKFILLED "priority" column, post-migration
    connection.close()  # => releases the connection immediately after reading
    return row  # => propagates the None case straight through to the caller

$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks/1
{"id":1,"title":"Buy milk","priority":3}
HTTP 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_pre_existing_row_reads_back_with_backfilled_priority PASSED [ 50%]
test_app.py::test_migration_mechanics_directly PASSED                    [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.19s =========================

"""Example 45: A Repository That Returns Typed Rows."""
 
# => co-24: this handler's own signature names TaskRow, so pyright checks the
#    SAME contract on both sides of the repository boundary, not just one
from fastapi import FastAPI  # => the web framework whose handler wraps the repository below
 
import repository  # => co-14: the module that owns EVERY database detail for this example
from repository import TaskRow  # => co-24: the typed shape this handler's own signature names
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, seeded tasks.db for every run
 
 
@app.get("/tasks")  # => co-08: a handler for listing tasks
def list_tasks() -> list[TaskRow]:  # => co-24: the handler's own signature names the typed shape
    tasks: list[TaskRow] = repository.list_tasks()  # => pyright checks this assignment for real
    total_title_chars = sum(  # => a trivial computation that only WORKS if "title" really exists
        len(task["title"])
        for task in tasks  # => typed access -- pyright knows "title" is a str
    )  # => typed access: pyright would flag task["missing_key"] as an error at edit time
    print(f"total title characters: {total_title_chars}")  # => proves the typed data is genuinely usable
    return tasks  # => TypedDict serializes to JSON exactly like a plain dict

"""Repository module for Example 45: typed return values."""
 
# => co-24: every function below returns TaskRow, never a bare dict or a raw
#    sqlite3.Row -- pyright can check the contract at the repository boundary
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => the ONLY database driver this module needs -- it ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
from typing import TypedDict  # => defines a precise, checkable dict shape
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
class TaskRow(TypedDict):  # => co-24: a precise, checkable shape instead of a bare dict
    id: int  # => pyright flags a missing or mistyped "id" key at edit time
    title: str  # => pyright flags a missing or mistyped "title" key at edit time
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not just row[0]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema and seeds two starter rows
    DB_PATH.unlink(missing_ok=True)  # => start every run from a clean, deterministic file
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape once, for the whole example
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => defines the table's shape once, for the whole example
    connection.executemany(  # => runs the SAME statement once per tuple below
        "INSERT INTO tasks (title) VALUES (?)",
        [("Buy milk",), ("Walk dog",)],  # => two seed rows
    )  # => two seed rows so list_tasks() below has something real to return
    connection.commit()  # => without this, the inserts never reach the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def _to_task_row(row: sqlite3.Row) -> TaskRow:  # => co-14: the ONE conversion point in this module
    # => co-14: the ONE place a raw sqlite3.Row becomes the typed TaskRow shape
    return TaskRow(id=row["id"], title=row["title"])  # => explicit keyword construction, type-checked
 
 
def list_tasks() -> list[TaskRow]:  # => co-24: the return type IS TaskRow, not sqlite3.Row
    connection = connect()  # => a fresh connection, scoped to just this call
    rows = connection.execute(  # => reads every row, oldest id first, as raw sqlite3.Row objects
        "SELECT id, title FROM tasks ORDER BY id"  # => no WHERE clause -- returns every row
    ).fetchall()  # => every row, oldest id first
    connection.close()  # => co-24: connection lifetime is scoped to one call, not shared state
    return [_to_task_row(row) for row in rows]  # => every caller gets TaskRow, never sqlite3.Row

$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks
[{"id":1,"title":"Buy milk"},{"id":2,"title":"Walk dog"}]
HTTP 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_typed_row_shape_is_preserved_over_json PASSED          [ 50%]
test_app.py::test_endpoint_returns_typed_rows_as_json PASSED             [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.15s =========================

"""Example 46: Layering -- No SQL in the Handler."""
 
# => co-24: this file is the PROOF -- its own test inspects this SOURCE FILE
#    and fails if any query keyword ever appears in it, not just at review time
from fastapi import FastAPI  # => the web framework whose handler wraps the repository below
 
import repository  # => co-24: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, seeded tasks.db for every run
 
 
@app.get("/tasks")  # => co-08: a handler for listing tasks
def list_tasks() -> list[dict[str, object]]:  # => a JSON array of plain dicts
    rows = repository.list_tasks()  # => co-24: the ENTIRE data-access line, one function call
    return [dict(row) for row in rows]  # => no query syntax, no "?", no sqlite3 import anywhere here

"""Repository module for Example 46: the only place SQL is allowed to live."""
 
# => co-24: every query keyword this whole example ever uses lives in THIS
#    file -- app.py's own test proves it by inspecting app.py's source text
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => the ONLY database driver this module needs -- it ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not just row[0]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema and seeds two starter rows
    DB_PATH.unlink(missing_ok=True)  # => start every run from a clean, deterministic file
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape once, for the whole example
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => defines the table's shape once, for the whole example
    connection.executemany(  # => runs the SAME statement once per tuple below
        "INSERT INTO tasks (title) VALUES (?)",
        [("Buy milk",), ("Walk dog",)],  # => two seed rows
    )  # => two seed rows so list_tasks() below has something real to return
    connection.commit()  # => without this, the inserts never reach the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def list_tasks() -> list[sqlite3.Row]:  # => co-24: the ONLY function in the example with a query
    # => co-24: this query is the ONLY data-access statement in the entire example --
    #    app.py never sees it, never imports sqlite3, never writes a query string
    connection = connect()  # => a fresh connection, scoped to just this call
    rows = connection.execute(  # => the one and only place this whole example reads data
        "SELECT id, title FROM tasks ORDER BY id"  # => no WHERE clause -- returns every row
    ).fetchall()  # => every row, oldest id first
    connection.close()  # => connection lifetime is scoped to one call, not shared state
    return rows  # => raw sqlite3.Row objects -- the handler converts them to dicts

$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks
[{"id":1,"title":"Buy milk"},{"id":2,"title":"Walk dog"}]
HTTP 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_app_module_contains_no_sql_keywords PASSED             [ 50%]
test_app.py::test_list_tasks_still_works_through_the_layered_repository PASSED [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.13s =========================

"""Example 47: FastAPI Depends() Supplies the DB Connection."""
 
# => co-23: the handler below never calls sqlite3.connect() or connection.close()
#    itself -- FastAPI's dependency injection owns the connection's entire lifecycle
import sqlite3  # => only needed for the type hint below, never for opening a connection
 
from fastapi import Depends, FastAPI  # => Depends() wires a dependency into a handler's signature
 
import repository  # => co-14/co-23: the module that owns both the SQL and the connection lifecycle
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, seeded tasks.db for every run
 
 
@app.get("/tasks")  # => co-08: a handler for listing tasks
def list_tasks_endpoint(
    connection: sqlite3.Connection = Depends(repository.get_connection),  # => co-23: injected per request
) -> list[dict[str, object]]:  # => a JSON array of plain dicts
    # => co-23: FastAPI calls get_connection() FOR THIS REQUEST and injects the
    #    value it yields here -- the handler never calls sqlite3.connect() itself
    rows = repository.list_tasks(connection)  # => co-14: SQL still lives only in the repository
    return [dict(row) for row in rows]  # => sqlite3.Row -> plain dict, JSON-serializable

"""Repository module for Example 47: SQL functions that accept an injected connection."""
 
# => co-23: get_connection() is a GENERATOR dependency -- FastAPI runs the code
#    before "yield" per request and GUARANTEES the "finally" cleanup runs after
from __future__ import annotations  # => lets sqlite3.Connection appear in signatures below
 
import sqlite3  # => the ONLY database driver this module needs -- it ships with Python itself
from collections.abc import Iterator  # => the return type a generator dependency needs
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def init_db() -> None:  # => (re)creates the schema and seeds two starter rows
    DB_PATH.unlink(missing_ok=True)  # => start every run from a clean, deterministic file
    connection = sqlite3.connect(DB_PATH)  # => a ONE-OFF connection, used only for setup
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not just row[0]
    connection.execute(  # => defines the table's shape once, for the whole example
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => defines the table's shape once, for the whole example
    connection.executemany(  # => runs the SAME statement once per tuple below
        "INSERT INTO tasks (title) VALUES (?)",
        [("Buy milk",), ("Walk dog",)],  # => two seed rows
    )  # => two seed rows so list_tasks() below has something real to return
    connection.commit()  # => without this, the inserts never reach the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def get_connection() -> Iterator[sqlite3.Connection]:  # => co-23: FastAPI drives this per request
    # => co-23: a GENERATOR dependency -- code before "yield" runs per request,
    #    code after "yield" runs once the response is done (guaranteed cleanup)
    connection = sqlite3.connect(DB_PATH)  # => one connection, opened fresh for THIS request
    connection.row_factory = sqlite3.Row  # => rows behave like dicts for every caller downstream
    try:  # => co-23: guarantees the connection closes even if the handler raises
        yield connection  # => the handler receives exactly THIS object as its argument
    finally:  # => runs on both the success path AND the exception path
        connection.close()  # => runs even if the handler raised an exception
 
 
def list_tasks(connection: sqlite3.Connection) -> list[sqlite3.Row]:  # => connection is a PARAMETER
    # => co-14: STILL the only place with a SELECT -- it just receives an
    #    already-open connection instead of opening its own, like earlier examples did
    return connection.execute(  # => no connect() call here -- the caller already supplied one
        "SELECT id, title FROM tasks ORDER BY id"  # => no WHERE clause -- returns every row
    ).fetchall()  # => every row, oldest id first -- no connect()/close() needed here

$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks
[{"id":1,"title":"Buy milk"},{"id":2,"title":"Walk dog"}]
HTTP 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_list_tasks_via_injected_connection PASSED              [ 50%]
test_app.py::test_get_connection_yields_and_then_closes PASSED           [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.21s =========================

"""Example 48: Middleware -- X-Request-Id."""  # => module docstring for this example
 
# => co-16: a middleware runs around EVERY route, once registered -- no
# => individual handler in this file ever mentions X-Request-Id itself
import uuid  # => stdlib UUID generation -- no third-party dependency needed for this
 
from fastapi import FastAPI, Request  # => Request is what every middleware receives first
from starlette.middleware.base import (  # => co-16: the two building blocks every middleware needs
    BaseHTTPMiddleware,  # => the base class FastAPI's own middleware system builds on
    RequestResponseEndpoint,  # => the precise type of "the rest of the pipeline", for typing call_next
)  # => closes the multi-line import from starlette.middleware.base
from starlette.responses import Response  # => the type dispatch() must always return
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
class RequestIdMiddleware(BaseHTTPMiddleware):  # => co-16: wraps EVERY request/response pair
    async def dispatch(  # => co-16: dispatch()'s signature spans three lines
        self,
        request: Request,
        call_next: RequestResponseEndpoint,  # => call_next runs the rest of the app
    ) -> Response:  # => must return the Response that eventually reaches the client
        response = await call_next(request)  # => runs routing + the matched handler first
        response.headers["X-Request-Id"] = str(  # => co-04: mutates the OUTGOING response headers
            uuid.uuid4()  # => a fresh, unpredictable id -- different on every single request
        )  # => co-04: a fresh id, stamped on the way back out
        return response  # => the SAME response object, now carrying the header
 
 
app.add_middleware(RequestIdMiddleware)  # => applies to EVERY route, with zero per-route code
 
# => registration order matters when multiple middlewares are stacked, but this
# => example registers only one, so ordering has no observable effect here
 
 
@app.get("/tasks")  # => co-08: a handler that knows NOTHING about request ids
def list_tasks() -> list[dict[str, str]]:  # => a plain handler, unaware any middleware even exists
    return [{"title": "Buy milk"}]  # => the middleware adds the header AFTER this returns

$ curl -s -D - -o /dev/null http://localhost:8000/tasks
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:01:16 GMT
server: uvicorn
content-length: 22
content-type: application/json
x-request-id: 80c3522c-a87a-42d9-8d96-630a8da3b317

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_response_carries_a_well_formed_request_id PASSED       [ 50%]
test_app.py::test_two_requests_get_two_different_request_ids PASSED      [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.15s =========================

"""Example 49: Middleware -- Request Logging."""  # => module docstring for this example
 
# => co-16: every request that reaches this app produces exactly ONE log
# => line, emitted here -- no individual handler ever calls logger.info() itself
import logging  # => Python's stdlib logging -- no third-party dependency needed
 
from fastapi import FastAPI, Request  # => Request is what every middleware receives first
from starlette.middleware.base import (  # => co-16: the two building blocks every middleware needs
    BaseHTTPMiddleware,  # => the base class FastAPI's own middleware system builds on
    RequestResponseEndpoint,  # => the precise type of "the rest of the pipeline", for typing call_next
)  # => closes the multi-line import from starlette.middleware.base
from starlette.responses import Response  # => the type dispatch() must always return
 
logging.basicConfig(level=logging.INFO)  # => ensures the "access" logger below is actually visible
logger = logging.getLogger("access")  # => co-16: a dedicated, named logger for this middleware
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
class LoggingMiddleware(BaseHTTPMiddleware):  # => co-16: wraps EVERY request/response pair
    async def dispatch(  # => co-16: dispatch()'s signature spans three lines
        self,  # => the middleware instance itself
        request: Request,  # => the incoming request, read but never mutated here
        call_next: RequestResponseEndpoint,  # => call_next runs the rest of the app
    ) -> Response:  # => must return the Response that eventually reaches the client
        response = await call_next(request)  # => the handler runs BEFORE the log line
        logger.info(  # => %s-style formatting -- logging module builds the string lazily
            "%s %s -> %s",  # => the format template: method, path, status
            request.method,  # => e.g. "GET"
            request.url.path,  # => e.g. "/tasks"
            response.status_code,  # => three fields, one line
        )  # => co-16: one structured line per request, method + path + outcome
        return response  # => the SAME response object, unmodified by logging
 
 
app.add_middleware(LoggingMiddleware)  # => applies to EVERY route, with zero per-route code
 
# => the "access" logger name is deliberate -- a real deployment could route it
# => to a separate log stream/file from application-level warnings and errors
 
 
@app.get("/tasks")  # => co-08: a handler that knows NOTHING about logging
def list_tasks() -> list[dict[str, str]]:  # => a plain handler, unaware any middleware even exists
    return [{"title": "Buy milk"}]  # => the middleware logs this request AFTER it returns

curl -s -o /dev/null http://localhost:8000/tasks

INFO:access:GET /tasks -> 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 1 item
 
test_app.py::test_request_produces_one_log_line PASSED                   [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 1 passed, 1 warning in 0.14s =========================

"""Example 50: Middleware -- Timing Header."""  # => module docstring for this example
 
# => co-16: the clock starts BEFORE call_next() and stops AFTER it, so the
# => elapsed time genuinely covers routing plus the handler, not just this wrapper
import time  # => stdlib high-resolution clock -- no third-party dependency needed
 
from fastapi import FastAPI, Request  # => Request is what every middleware receives first
from starlette.middleware.base import (  # => co-16: the two building blocks every middleware needs
    BaseHTTPMiddleware,  # => the base class FastAPI's own middleware system builds on
    RequestResponseEndpoint,  # => the precise type of "the rest of the pipeline", for typing call_next
)  # => closes the multi-line import from starlette.middleware.base
from starlette.responses import Response  # => the type dispatch() must always return
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
class TimingMiddleware(BaseHTTPMiddleware):  # => co-16: wraps EVERY request/response pair
    async def dispatch(  # => co-16: dispatch()'s signature spans three lines
        self,
        request: Request,
        call_next: RequestResponseEndpoint,  # => call_next runs the rest of the app
    ) -> Response:  # => must return the Response that eventually reaches the client
        start = time.perf_counter()  # => co-16: wall-clock start, BEFORE the handler runs
        response = await call_next(request)  # => the handler (and any inner middleware) runs here
        elapsed = time.perf_counter() - start  # => elapsed time AFTER the handler returned
        response.headers["X-Process-Time"] = (  # => co-04: mutates the OUTGOING response headers
            f"{elapsed:.6f}"  # => co-04: seconds, formatted as a string response header
        )  # => closes the header assignment
        return response  # => the SAME response object, now carrying the timing header
 
 
app.add_middleware(TimingMiddleware)  # => applies to EVERY route, with zero per-route code
 
# => perf_counter(), not time.time() -- a monotonic clock immune to system
# => clock adjustments, which is what any real timing measurement needs
 
 
@app.get("/tasks")  # => co-08: a handler that knows NOTHING about timing
def list_tasks() -> list[dict[str, str]]:  # => a plain handler, unaware any middleware even exists
    return [{"title": "Buy milk"}]  # => the middleware measures this AFTER it returns

$ curl -s -D - -o /dev/null http://localhost:8000/tasks
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:01:25 GMT
server: uvicorn
content-length: 22
content-type: application/json
x-process-time: 0.001788

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 1 item
 
test_app.py::test_response_carries_a_parseable_timing_header PASSED      [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 1 passed, 1 warning in 0.15s =========================

"""Example 51: Middleware -- CORS."""
 
# => co-16: unlike ex-48/49/50, this middleware ships WITH FastAPI -- configured
#    once here, not hand-written, and still applies to every route below
from fastapi import FastAPI  # => the web framework whose built-in CORS middleware this example uses
from fastapi.middleware.cors import CORSMiddleware  # => co-16: ships with FastAPI, no hand-rolling needed
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
app.add_middleware(  # => co-16: FastAPI's OWN built-in middleware, not hand-rolled
    CORSMiddleware,  # => the middleware class being registered
    allow_origins=["https://example.com"],  # => co-04: only THIS origin gets the header back
    allow_methods=["GET"],  # => only GET is permitted cross-origin for this app
)  # => closes the add_middleware(...) call
 
 
@app.get("/tasks")  # => co-08: a handler that knows NOTHING about CORS
def list_tasks() -> list[dict[str, str]]:  # => a plain handler, unaware any middleware even exists
    return [{"title": "Buy milk"}]  # => CORSMiddleware decides the header, not this function

$ curl -s -D - -o /dev/null -H "Origin: https://example.com" http://localhost:8000/tasks
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:01:26 GMT
server: uvicorn
content-length: 22
content-type: application/json
access-control-allow-origin: https://example.com
vary: Origin
 
 
$ curl -s -D - -o /dev/null -H "Origin: https://evil.example" http://localhost:8000/tasks
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:01:26 GMT
server: uvicorn
content-length: 22
content-type: application/json

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_allowed_origin_gets_cors_header PASSED                 [ 50%]
test_app.py::test_disallowed_origin_gets_no_cors_header PASSED           [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.15s =========================

"""Example 52: A Sanitized 500 Envelope for Unhandled Exceptions."""
 
# => co-11: the /boom handler below ALWAYS raises -- real uvicorn+curl gets
#    the sanitized JSON below; the raw exception detail only ever reaches server logs
from fastapi import FastAPI, Request  # => Request gives the handler access to the raw ASGI request
from fastapi.responses import JSONResponse  # => the response type this handler builds by hand
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
@app.exception_handler(Exception)  # => co-11: catches EVERYTHING not already handled elsewhere
async def unhandled_exception_handler(
    request: Request,
    exc: Exception,  # => exc is the ORIGINAL exception, never shown to the caller
) -> JSONResponse:  # => must return a Response FastAPI can send back to the client
    # => deliberately generic -- never leaks exc's message or a stack trace to the client;
    #    a real deployment would log str(exc) SERVER-SIDE here, just never return it
    return JSONResponse(  # => builds the sanitized envelope BY HAND, replacing the raw traceback
        status_code=500,  # => co-03: server-side failure
        content={  # => the JSON body FastAPI serializes and sends back to the client
            "error": {  # => nests everything under ONE top-level key, matching earlier envelopes
                "code": "internal_error",  # => a stable, machine-matchable code
                "message": "an unexpected error occurred",  # => deliberately generic, every time
            }  # => closes the inner "error" object
        },  # => closes the "content" dict passed to JSONResponse
    )  # => closes the JSONResponse(...) call itself
 
 
@app.get("/boom")  # => co-08: a handler that always fails, on purpose, for this example
def boom() -> None:  # => never returns normally -- exists solely to trigger the handler above
    raise RuntimeError(  # => co-11: an ORDINARY, unhandled exception -- not a domain error
        "something exploded with sensitive internal details"  # => never reaches the client
    )  # => this string never reaches the client -- the handler above replaces it entirely

$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/boom
{"error":{"code":"internal_error","message":"an unexpected error occurred"}}
HTTP 500

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 1 item
 
test_app.py::test_unhandled_exception_returns_sanitized_500 PASSED       [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 1 passed, 1 warning in 0.13s =========================

"""Example 53: The 422 Detail Array Lists Every Offending Field."""
 
# => the SAME TaskCreate shape as ex-31, but this example's test specifically
#    breaks BOTH fields at once to show the detail array grows, not just the first
from fastapi import FastAPI  # => the web framework whose validation layer this example exercises
from pydantic import BaseModel, Field  # => Field() attaches constraints beyond a bare type
 
app = FastAPI()  # => the ASGI application uvicorn will serve
 
 
class TaskCreate(BaseModel):  # => the shape of a valid POST /tasks body
    title: str = Field(min_length=1)  # => two INDEPENDENT constraints...
    priority: int = Field(gt=0)  # => ...that can both fail on the SAME request
 
 
@app.post("/tasks", status_code=201)  # => co-08: a handler for creating a task
def create_task(task: TaskCreate) -> dict[str, object]:  # => "task" only exists if BOTH fields passed
    return {"title": task.title, "priority": task.priority}  # => both fields, already constraint-checked

$ curl -s -w "\nHTTP %{http_code}\n" -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{"title": "", "priority": 0}'
{"detail":[{"type":"string_too_short","loc":["body","title"],"msg":"String should have at least 1 character","input":"","ctx":{"min_length":1}},{"type":"greater_than","loc":["body","priority"],"msg":"Input should be greater than 0","input":0,"ctx":{"gt":0}}]}
HTTP 422

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_single_violation_yields_one_detail_entry PASSED        [ 50%]
test_app.py::test_two_simultaneous_violations_yield_two_detail_entries PASSED [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 2 passed, 1 warning in 0.21s =========================

"""Example 54: Hand-Written Accept Header Negotiation."""  # => module docstring for this example
 
from fastapi import Depends, FastAPI, Header, HTTPException  # => Header() reads a request header
 
app = FastAPI()  # => the ASGI application uvicorn will serve
# => co-21: FastAPI does NOT enforce `Accept` negotiation on its own. Its built-in
# => strict_content_type_checking only validates the REQUEST's Content-Type; nothing
# => in the framework inspects `Accept` -- the dependency below is entirely hand-written
 
 
def require_json_accept(accept: str = Header(default="*/*")) -> None:  # => co-21: a FastAPI dependency
    # => co-21: a DEPENDENCY that runs BEFORE the handler -- accepts "application/json"
    # => (or any wildcard that matches it) and rejects everything else with a 406
    if "application/json" not in accept and "*/*" not in accept:  # => the entire negotiation rule
        raise HTTPException(status_code=406, detail="only application/json is supported")  # => co-21: 406
 
 
@app.get("/tasks", dependencies=[Depends(require_json_accept)])  # => runs on EVERY call to this route
def list_tasks() -> list[dict[str, str]]:  # => a plain handler, reached only after the dependency passes
    return [{"title": "Buy milk"}]  # => only reached once require_json_accept() did not raise

$ curl -s -w "\nHTTP %{http_code}\n" -H "Accept: application/json" http://localhost:8000/tasks
[{"title":"Buy milk"}]
HTTP 200
 
$ curl -s -w "\nHTTP %{http_code}\n" -H "Accept: text/html" http://localhost:8000/tasks
{"detail":"only application/json is supported"}
HTTP 406

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_explicit_json_accept_is_allowed PASSED                 [ 33%]
test_app.py::test_wildcard_accept_is_allowed PASSED                      [ 66%]
test_app.py::test_unsupported_accept_is_rejected_with_406 PASSED         [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 3 passed, 1 warning in 0.17s =========================

"""Example 55: Create Then Read -- A Persisted Round Trip."""
 
# => co-02: POST writes, then a SEPARATE GET call reads the SAME id back --
#    proving the write genuinely reached durable storage, not just an in-memory echo
from fastapi import FastAPI, HTTPException  # => HTTPException raises FastAPI's default error shape
from pydantic import BaseModel  # => Pydantic models are FastAPI's validation vocabulary
 
import repository  # => co-14: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn will serve
repository.init_db()  # => fresh, empty tasks.db for every run
 
 
class TaskCreate(BaseModel):  # => the shape of a valid POST /tasks body
    title: str  # => co-10: validated before either handler below sees it
 
 
@app.post("/tasks", status_code=201)  # => co-02: creates a new resource
def create_task(task: TaskCreate) -> dict[str, object]:  # => "task" only exists if validation passed
    new_id = repository.create_task(task.title)  # => step one of the round trip: the write
    return {"id": new_id, "title": task.title}  # => echoes the id the DB actually assigned
 
 
@app.get("/tasks/{task_id}")  # => co-02: reading the SAME resource back afterward
def get_task(task_id: int) -> dict[str, object]:  # => task_id arrives already parsed as an int
    row = repository.get_task(task_id)  # => step two of the round trip: the read
    if row is None:  # => the ONLY branch this handler makes -- everything else lives in the repo
        raise HTTPException(status_code=404)  # => FastAPI's plain default 404 shape
    return dict(row)  # => co-14: proves the POST actually reached durable storage

"""Repository module for Example 55: create then read, in one round trip."""
 
# => co-02/co-14: create_task() and get_task() together prove the WRITE
#    genuinely reaches the SAME file the next call's READ opens, not just memory
from __future__ import annotations  # => lets sqlite3.Row appear in return-type hints below
 
import sqlite3  # => the ONLY database driver this module needs -- it ships with Python itself
from pathlib import Path  # => builds an absolute, OS-independent path to the db file
 
DB_PATH = Path(__file__).parent / "tasks.db"  # => co-14: one fixed db file, next to this module
 
 
def connect() -> sqlite3.Connection:  # => opens and configures one sqlite3 connection
    connection = sqlite3.connect(DB_PATH)  # => DB_PATH is the single file this module reads/writes
    connection.row_factory = sqlite3.Row  # => rows behave like dicts: row["title"], not just row[0]
    return connection  # => the caller owns closing this connection when done
 
 
def init_db() -> None:  # => (re)creates the schema, starting EMPTY
    DB_PATH.unlink(missing_ok=True)  # => start every run from a clean, deterministic file
    connection = connect()  # => a fresh connection, scoped to just this setup call
    connection.execute(  # => defines the table's shape -- no seed rows this time
        "CREATE TABLE tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)"  # => the exact DDL
    )  # => starts EMPTY -- this example is entirely about the create-then-read round trip
    connection.commit()  # => without this, the table definition never reaches the file on disk
    connection.close()  # => releases the connection -- init_db() is a one-shot setup call
 
 
def create_task(title: str) -> int:  # => co-02/co-14: step one of the round trip -- the write half
    # => co-02/co-14: step one of the round trip -- the write half
    connection = connect()  # => a fresh connection, scoped to just this call
    cursor = connection.execute(  # => "?" binds title as DATA, the same safety property as before
        "INSERT INTO tasks (title) VALUES (?)",
        (title,),  # => one placeholder, one bound value
    )  # => "?" binds title as DATA, the same safety property as every earlier example
    connection.commit()  # => without this, the row exists only in this connection's transaction
    new_id = cursor.lastrowid  # => sqlite3 exposes the AUTOINCREMENT id straight off the cursor
    connection.close()  # => releases the connection immediately after writing
    assert new_id is not None  # => guaranteed non-None right after a successful INSERT
    return new_id  # => the id the read half will use to prove persistence
 
 
def get_task(task_id: int) -> sqlite3.Row | None:  # => co-02/co-14: step two of the round trip
    # => co-02/co-14: step two of the round trip -- the read half
    connection = connect()  # => a fresh connection, scoped to just this call
    row = connection.execute(  # => a SEPARATE connection from the one create_task() used
        "SELECT id, title FROM tasks WHERE id = ?",
        (task_id,),  # => one placeholder, one bound value
    ).fetchone()  # => a single Row, if the earlier INSERT genuinely persisted
    connection.close()  # => co-24: connection lifetime is scoped to one call, not shared state
    return row  # => propagates the None case straight through to the caller

$ curl -s -w "\nHTTP %{http_code}\n" -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{"title": "Buy milk"}'
{"id":1,"title":"Buy milk"}
HTTP 201
 
$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks/1
{"id":1,"title":"Buy milk"}
HTTP 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 1 item
 
test_app.py::test_create_then_read_round_trip PASSED                     [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 1 passed, 1 warning in 0.14s =========================

"""Example 56: pytest + FastAPI's TestClient."""
 
# => co-22: this app.py is unremarkable on purpose -- the interesting part of
#    this example lives entirely in test_app.py's fixture, not in this handler code
from fastapi import FastAPI  # => the web framework whose handlers this example's tests exercise
from pydantic import BaseModel  # => Pydantic models are FastAPI's validation vocabulary
 
import repository  # => co-14: the module that owns EVERY database detail for this example
 
app = FastAPI()  # => the ASGI application uvicorn (and TestClient) will serve
repository.init_db()  # => a first, module-import-time reset -- the pytest fixture resets it AGAIN per test
 
 
class TaskCreate(BaseModel):  # => the shape of a valid POST /tasks body
    title: str  # => co-10: validated before either handler below sees it
 
 
@app.post("/tasks", status_code=201)  # => co-08: a handler for creating a task
def create_task(task: TaskCreate) -> dict[str, object]:  # => "task" only exists if validation passed
    new_id = repository.create_task(task.title)  # => co-14: the handler delegates the INSERT
    return {"id": new_id, "title": task.title}  # => echoes the id the DB actually assigned
 
 
@app.get("/tasks")  # => co-14: returns a JSON ARRAY, one element per row
def list_tasks() -> list[dict[str, object]]:  # => the array shape a client can iterate directly
    return [dict(row) for row in repository.list_tasks()]  # => sqlite3.Row -> plain dict, per row

"""Tests for Example 56: pytest + FastAPI's TestClient."""
 
# => co-22: the fixture below is the WHOLE point of this example -- every
#    test function receives a freshly reset TestClient, never a shared one
from collections.abc import Iterator  # => the precise return type a generator fixture needs
 
import pytest  # => co-22: pytest's own fixture/test discovery machinery
import repository  # => co-14: the module whose init_db() resets state per test
from fastapi.testclient import TestClient  # => co-22: drives the ASGI app without a real socket
 
from app import app  # => the SAME FastAPI instance uvicorn would otherwise serve
 
 
@pytest.fixture()  # => co-22: pytest calls this ONCE PER TEST FUNCTION that requests it
def client() -> Iterator[TestClient]:  # => a generator fixture -- code after yield is teardown
    # => co-22: TestClient drives the ASGI app IN-PROCESS -- no uvicorn, no real
    #    socket, no network round trip -- yet it exercises the exact same FastAPI
    #    routing/validation/middleware stack a real HTTP request would go through
    repository.init_db()  # => a FRESH database before every single test function
    with TestClient(app) as test_client:  # => "with" also runs startup/shutdown events
        yield test_client  # => each test function receives THIS client as its argument
 
 
def test_create_task_returns_201(client: TestClient) -> None:  # => "client" is the fixture above
    response = client.post("/tasks", json={"title": "Buy milk"})  # => in-process POST, no curl needed
    assert response.status_code == 201  # => co-03: matches the app's status_code=201 decorator arg
    assert response.json()["title"] == "Buy milk"  # => the body round-trips through real validation
 
 
def test_list_tasks_reflects_created_tasks(client: TestClient) -> None:  # => a fresh client, again
    client.post("/tasks", json={"title": "Buy milk"})  # => grows the list this test's own client sees
    client.post("/tasks", json={"title": "Walk dog"})  # => a second row, same isolated database
    response = client.get("/tasks")  # => reads back everything this test itself just wrote
    assert response.status_code == 200  # => the plain success case
    titles = [task["title"] for task in response.json()]  # => extracts just the field being asserted
    assert titles == ["Buy milk", "Walk dog"]  # => insertion order, proven end to end
 
 
def test_each_test_gets_an_isolated_database(client: TestClient) -> None:  # => the ISOLATION proof
    # => proves the fixture's repository.init_db() call genuinely resets state --
    #    if a previous test's rows had leaked in, this list would NOT be empty
    response = client.get("/tasks")  # => runs AFTER the two tests above, in whatever order pytest picks
    assert response.status_code == 200  # => the endpoint itself never fails
    assert response.json() == []  # => this test creates nothing, so the list starts empty

$ curl -s -w "\nHTTP %{http_code}\n" -X POST http://localhost:8000/tasks -H "Content-Type: application/json" -d '{"title": "Buy milk"}'
{"id":1,"title":"Buy milk"}
HTTP 201
 
$ curl -s -w "\nHTTP %{http_code}\n" http://localhost:8000/tasks
[{"id":1,"title":"Buy milk"}]
HTTP 200

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_create_task_returns_201 PASSED                         [ 33%]
test_app.py::test_list_tasks_reflects_created_tasks PASSED               [ 66%]
test_app.py::test_each_test_gets_an_isolated_database PASSED             [100%]
 
=============================== warnings summary ===============================
../../../.venv/lib/python3.13/site-packages/fastapi/testclient.py:1
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================= 3 passed, 1 warning in 0.15s =========================