Advanced Examples

"""Example 57: Sessions vs Tokens -- two ways to identify the same caller."""
# => co-17: both mechanisms below answer the SAME question ("who is this caller?") with
# => two different tradeoffs -- session state lives on the server, token state lives on the wire
 
import uuid  # => generates unpredictable session ids -- a real server never uses a guessable id
 
from fastapi import Cookie, FastAPI, Header, HTTPException, Response  # => co-04: header/cookie params
 
app = FastAPI()  # => a fresh app; this example needs no persistence at all
 
# -- Session mechanism: server holds the state (co-17) -----------------------------------
SESSIONS: dict[str, str] = {}  # => session_id -> username; lives ONLY in this process's memory
# => a real deployment would put this in Redis/DB precisely because in-process memory
# => does not survive a restart or scale past one worker (co-05 tension, revisited ex-80)
 
# -- Token mechanism: the token itself IS the credential (co-17) -------------------------
VALID_TOKEN = "s3cr3t-token-abc123"  # => a hardcoded stand-in for a real signed/opaque token
# => (real tokens are cryptographically signed or opaque-and-looked-up -- never a literal, guessable
# => string -- this constant exists purely so curl can present a value the server will accept)
TOKEN_USER = "alice"  # => the token's fixed "owner" for this pedagogical example
 
 
@app.post("/login-session")  # => co-17: issues a SESSION -- server-side state plus a cookie
def login_session(response: Response) -> dict[str, str]:
    session_id = str(uuid.uuid4())  # => a fresh, unguessable identifier for this session
    SESSIONS[session_id] = "alice"  # => the SERVER remembers who this session belongs to
    response.set_cookie("session_id", session_id)  # => co-04: the CLIENT only gets an opaque id
    return {"mechanism": "session", "session_id": session_id}  # => confirms issuance to the caller
 
 
@app.get("/profile-session")  # => co-17: identifies the caller via the session cookie
def profile_session(session_id: str | None = Cookie(default=None)) -> dict[str, str]:
    if session_id is None or session_id not in SESSIONS:  # => no cookie, or server forgot it
        raise HTTPException(status_code=401, detail="no valid session")  # => co-03: 401 unauthenticated
    username = SESSIONS[session_id]  # => O(1) lookup -- but ONLY works on the process that has it
    return {"mechanism": "session", "username": username}  # => the caller's identity, resolved
 
 
@app.post("/login-token")  # => co-17, co-18: issues a TOKEN -- no server-side state created
def login_token() -> dict[str, str]:
    return {"mechanism": "token", "token": VALID_TOKEN}  # => the token itself carries everything needed
 
 
@app.get("/profile-token")  # => co-17, co-18: identifies the caller via the bearer token alone
def profile_token(authorization: str | None = Header(default=None)) -> dict[str, str]:
    if authorization != f"Bearer {VALID_TOKEN}":  # => co-04: reads the raw Authorization header
        raise HTTPException(status_code=401, detail="missing or invalid token")  # => 401, same as session
    return {"mechanism": "token", "username": TOKEN_USER}  # => resolved WITHOUT any server-side lookup

=== POST /login-session ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:54:32 GMT
server: uvicorn
content-length: 75
content-type: application/json
set-cookie: session_id=cb7cdbbd-0544-4ed7-9364-89c5de3f101b; Path=/; SameSite=lax
 
{"mechanism":"session","session_id":"cb7cdbbd-0544-4ed7-9364-89c5de3f101b"}
=== GET /profile-session (with cookie) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:54:32 GMT
server: uvicorn
content-length: 42
content-type: application/json
 
{"mechanism":"session","username":"alice"}
=== GET /profile-session (no cookie) ===
HTTP/1.1 401 Unauthorized
date: Tue, 14 Jul 2026 10:54:32 GMT
server: uvicorn
content-length: 29
content-type: application/json
 
{"detail":"no valid session"}
=== POST /login-token ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:54:32 GMT
server: uvicorn
content-length: 51
content-type: application/json
 
{"mechanism":"token","token":"s3cr3t-token-abc123"}
=== GET /profile-token (valid) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:54:32 GMT
server: uvicorn
content-length: 40
content-type: application/json
 
{"mechanism":"token","username":"alice"}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 4 items
 
test_app.py::test_session_login_sets_cookie_and_profile_reads_it PASSED  [ 25%]
test_app.py::test_profile_session_without_cookie_is_401 PASSED           [ 50%]
test_app.py::test_token_login_and_profile_identify_caller PASSED         [ 75%]
test_app.py::test_profile_token_without_header_is_401 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
========================= 4 passed, 1 warning in 0.16s =========================

"""Example 58: Issue a Token -- POST /login returns a token string."""
# => co-17: this is the OTHER half of Example 57's token mechanism -- where does the token
# => a client presents on every later call actually come FROM? This is that issuing endpoint.
 
from fastapi import FastAPI, HTTPException  # => co-17, co-18: raises 401 on bad credentials
from pydantic import BaseModel  # => co-10: a typed model validates the login body
 
app = FastAPI()  # => a fresh app -- no persistence, no session store, nothing survives a restart
 
VALID_TOKEN = "s3cr3t-token-abc123"  # => hardcoded stand-in for a real signed/opaque token
# => (a real login endpoint would look up the user, verify a hashed password, then mint a
# => fresh signed JWT or opaque token -- this example fixes ALL of that to one literal string
# => so the FOCUS stays on the request/response SHAPE, not on cryptography or password hashing)
USERNAME = "alice"  # => the ONE registered user this pedagogical example recognizes
PASSWORD = "wonderland"  # => a hardcoded password -- never do this in real code (co-17 caveat)
 
 
class Credentials(BaseModel):  # => co-10: the shape POST /login expects
    username: str  # => a plain required string field
    password: str  # => a plain required string field, never returned in any response
    # => (a real API would also never LOG this field -- pydantic gives no free redaction,
    # => so that discipline has to be a deliberate choice in whatever logs the request)
 
 
class TokenResponse(BaseModel):  # => co-09: declares the exact shape of a successful login
    token: str  # => the ONLY thing the client needs to authenticate future requests
 
 
@app.post("/login", response_model=TokenResponse)  # => co-17, co-18: issues a token, no session created
def login(credentials: Credentials) -> TokenResponse:
    # => co-10: FastAPI already validated `credentials` matches the Credentials shape above --
    # => by the time this line runs, both fields are guaranteed to be present strings
    if credentials.username != USERNAME or credentials.password != PASSWORD:  # => co-10: reject any mismatch immediately -- deliberately the SAME message for both
        # => a wrong username and a wrong password, so a caller can't enumerate valid usernames
        raise HTTPException(status_code=401, detail="invalid credentials")  # => co-03: 401, not 404
    return TokenResponse(token=VALID_TOKEN)  # => co-09: the response body IS just the token --
    # => no session id, no server-side record created -- co-17's defining difference from ex-57

=== POST /login (correct credentials) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:54:58 GMT
server: uvicorn
content-length: 31
content-type: application/json
 
{"token":"s3cr3t-token-abc123"}
=== POST /login (wrong password) ===
HTTP/1.1 401 Unauthorized
date: Tue, 14 Jul 2026 10:54:58 GMT
server: uvicorn
content-length: 32
content-type: application/json
 
{"detail":"invalid credentials"}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_login_with_correct_credentials_returns_token PASSED    [ 33%]
test_app.py::test_login_with_wrong_password_is_401 PASSED                [ 66%]
test_app.py::test_login_missing_field_is_422 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 59: Token-Check Middleware -- validates a Bearer token on protected routes."""
# => co-16: a MIDDLEWARE runs on every request BEFORE routing decides which handler runs --
# => that makes it the right place for a cross-cutting concern like "is this caller allowed in?"
# => rather than repeating the same check inside every individual handler function
 
from collections.abc import Awaitable, Callable  # => precise typing for the middleware's signature
 
from fastapi import FastAPI, Request, Response  # => co-16: middleware operates on raw Request/Response
from fastapi.responses import JSONResponse  # => co-11: builds the structured 401 body by hand
 
app = FastAPI()  # => a fresh app -- this example needs no database, only routing + middleware
 
VALID_TOKEN = "s3cr3t-token-abc123"  # => hardcoded stand-in for a real signed/opaque token
 
 
@app.middleware("http")  # => co-16: registers a function that wraps EVERY request/response --
# => this decorator form is Starlette's original middleware API; FastAPI also accepts
# => `app.add_middleware(...)` for class-based middleware, but a function is simplest here
async def token_check_middleware(
    request: Request,  # => co-16: the INCOMING request -- path, method, and headers are readable here
    call_next: Callable[[Request], Awaitable[Response]],  # => co-16: calling this continues the chain
) -> Response:
    # => co-16: this `if` is the ENTIRE authorization policy for the whole app in one place --
    # => change what "needs a token" means by editing this one condition, not N handler functions
    if request.url.path.startswith("/protected"):  # => co-16: only THIS path prefix is guarded --
        # => Example 63 explores a read/write split instead of a path-prefix split like this one
        auth_header = request.headers.get("authorization")  # => co-04: reads the raw header --
        # => note the lowercase key: HTTP header names are case-insensitive, and Starlette's
        # => Headers mapping normalizes lookups accordingly regardless of the wire casing
        if auth_header != f"Bearer {VALID_TOKEN}":  # => co-18: exact string match, no partial credit --
            # => this branch fires for BOTH a missing header (None != "Bearer ...") and a garbage one
            return JSONResponse(  # => co-11: a structured envelope, never a stack trace or bare text
                status_code=401,  # => co-03: 401 Unauthorized -- "who you claim to be" was rejected
                content={"error": {"code": "unauthorized", "message": "missing or invalid token"}},
            )  # => co-16: returning HERE short-circuits the chain -- call_next() never runs,
            # => so the guarded handler function below never even starts executing
        request.state.user = "alice"  # => co-16: a valid token attaches identity for the handler to use --
        # => request.state is a plain per-request namespace; middleware writes to it, handlers read it
    return await call_next(request)  # => co-16: hands off to routing -- unmodified for open paths,
    # => and for /protected paths that passed the check above
 
 
@app.get("/public")  # => never touches the middleware's token branch at all -- doesn't start with /protected
def public() -> dict[str, str]:
    return {"access": "public"}  # => reachable with zero headers, zero token, always
 
 
@app.get("/protected/data")  # => co-18: only reachable with a valid token, per the middleware above --
# => this handler body itself contains NO auth logic whatsoever -- that separation is the whole point
def protected_data(request: Request) -> dict[str, str]:
    return {"access": "protected", "user": request.state.user}  # => identity set by the middleware,
    # => not re-derived here -- the handler simply trusts what already ran before it

=== GET /public (no token needed) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:55:00 GMT
server: uvicorn
content-length: 19
content-type: application/json
 
{"access":"public"}
=== GET /protected/data (valid token) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:55:00 GMT
server: uvicorn
content-length: 37
content-type: application/json
 
{"access":"protected","user":"alice"}
=== GET /protected/data (no token) ===
HTTP/1.1 401 Unauthorized
date: Tue, 14 Jul 2026 10:55:00 GMT
server: uvicorn
content-length: 70
content-type: application/json
 
{"error":{"code":"unauthorized","message":"missing or invalid token"}}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_public_route_needs_no_token PASSED                     [ 33%]
test_app.py::test_protected_route_with_valid_token_reaches_handler PASSED [ 66%]
test_app.py::test_protected_route_without_token_is_401 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 60: Missing Token -- a protected route with no Authorization header at all."""
# => co-18: this is the FIRST of three 401 scenarios (60, 61, 62) that all reuse the SAME
# => dependency below -- only the request curl sends differs, which is the whole teaching point
 
from fastapi import Depends, FastAPI, HTTPException, Request  # => co-23: Depends injects the auth check
from fastapi.responses import JSONResponse  # => co-11: builds the exception handler's structured body
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer  # => co-18: standard bearer scheme
 
app = FastAPI()  # => a fresh app -- this example needs no database, only auth plumbing
 
 
@app.exception_handler(HTTPException)  # => co-11: UNWRAPS FastAPI's default {"detail": ...} nesting --
# => without this handler, raising HTTPException(detail={"error": {...}}) would ship as
# => {"detail": {"error": {...}}}, doubly-nested and inconsistent with every OTHER error path
async def structured_error_handler(request: Request, exc: HTTPException) -> JSONResponse:
    body = (
        exc.detail
        if isinstance(exc.detail, dict)  # => the handlers below always raise dict-shaped detail
        else {"error": {"code": "error", "message": str(exc.detail)}}  # => fallback for plain-string detail
    )  # => reuses the raised dict AS-IS -- one consistent {"error": {...}} shape, no extra nesting
    return JSONResponse(status_code=exc.status_code, content=body)  # => co-11: same envelope, every error
 
 
VALID_TOKEN = "s3cr3t-token-abc123"  # => hardcoded stand-in for a real signed/opaque token
security = HTTPBearer(auto_error=False)  # => auto_error=False: WE decide the error shape, not
# => FastAPI's own default (which would raise a bare 403 with a generic "Not authenticated" body)
 
 
def require_token(  # => co-23: a reusable DEPENDENCY -- any route can opt in via Depends(require_token)
    credentials: HTTPAuthorizationCredentials | None = Depends(security),  # => co-23: FastAPI calls
    # => `security` FIRST, parses any Authorization header it finds, and hands the RESULT here
) -> str:
    if credentials is None:  # => co-18: no Authorization header was sent AT ALL -- this is THIS
        # => example's scenario specifically: curl with no -H "Authorization: ..." flag whatsoever
        raise HTTPException(  # => co-11: structured envelope, not a bare string
            status_code=401,  # => co-03: 401 -- the caller never even attempted to identify itself
            detail={"error": {"code": "unauthorized", "message": "Authorization header missing"}},
        )
    if credentials.credentials != VALID_TOKEN:  # => header present, but the token itself is wrong --
        # => Example 61 is the one that actually exercises THIS branch; unreachable from THIS example's curl
        raise HTTPException(
            status_code=401,
            detail={"error": {"code": "unauthorized", "message": "invalid token"}},
        )
    return "alice"  # => the resolved identity, available to any handler that depends on this --
    # => Example 62 is the one that actually reaches this successful return
 
 
@app.get("/protected")  # => co-18: guarded entirely by the require_token dependency above --
# => this handler's body contains ZERO auth logic; FastAPI runs the dependency before this even starts
def protected(user: str = Depends(require_token)) -> dict[str, str]:
    # => co-23: by the time execution reaches THIS line, `user` is guaranteed to be "alice" --
    # => every path that could have failed already returned a 401 response one call frame up
    return {"user": user}  # => only reached once require_token returns without raising

=== GET /protected (no Authorization header at all) ===
HTTP/1.1 401 Unauthorized
date: Tue, 14 Jul 2026 10:55:03 GMT
server: uvicorn
content-length: 74
content-type: application/json
 
{"error":{"code":"unauthorized","message":"Authorization header missing"}}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_no_authorization_header_is_401_with_missing_message PASSED [ 50%]
test_app.py::test_valid_token_still_reaches_handler_for_contrast 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 61: Invalid Token -- a protected route with a malformed/wrong token."""
# => co-18: a token can be WRONG in two different ways -- entirely absent (Example 60) or
# => present-but-incorrect (this example) -- and a caller sees the SAME 401 status either way,
# => which is deliberate: never leak which half of "no token" vs "bad token" a caller hit
 
from fastapi import Depends, FastAPI, HTTPException, Request  # => co-18: same dependency style as ex-60
from fastapi.responses import JSONResponse  # => co-11: builds the exception handler's structured body
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer  # => co-18: parses "Bearer <token>"
 
app = FastAPI()  # => a fresh app -- this example needs no database, only auth plumbing
# => (fully self-contained: nothing here is imported from any other example directory)
 
VALID_TOKEN = "s3cr3t-token-abc123"  # => hardcoded stand-in for a real signed/opaque token
security = HTTPBearer(auto_error=False)  # => auto_error=False: WE own the 401 body's shape below,
# => not FastAPI's own built-in "Not authenticated" 403 default
# => (co-18: 401 means "I don't know who you are"; 403 means "I know, and you're not allowed")
 
 
@app.exception_handler(HTTPException)  # => co-11: one consistent {"error": {...}} envelope, unwrapped --
# => without this, every raise HTTPException(detail={"error": {...}}) below would ship doubly
# => nested as {"detail": {"error": {...}}} instead of the flat {"error": {...}} shape shown here
async def structured_error_handler(request: Request, exc: HTTPException) -> JSONResponse:
    body = (
        exc.detail
        if isinstance(exc.detail, dict)  # => every raise below already supplies a dict -- this
        # => isinstance check exists so the handler ALSO degrades gracefully for a plain string detail
        else {"error": {"code": "error", "message": str(exc.detail)}}
    )
    return JSONResponse(status_code=exc.status_code, content=body)  # => co-11: same shape, every error
 
 
def require_token(  # => co-23: identical dependency SHAPE to Example 60 -- the difference this
    # => example actually exercises lives entirely in what curl sends, not in this function's code
    credentials: HTTPAuthorizationCredentials | None = Depends(security),
) -> str:
    if credentials is None:  # => genuinely absent -- Example 60's case, still handled here for completeness
        # => so this dependency stays a fully correct, standalone, self-contained auth check on its own
        raise HTTPException(
            status_code=401,
            detail={"error": {"code": "unauthorized", "message": "Authorization header missing"}},
        )
    if credentials.credentials != VALID_TOKEN:  # => co-18: THIS example's focus -- present but WRONG --
        raise HTTPException(  # => a well-formed "Bearer <garbage>" header still fails the equality check;
            # => HTTPBearer already stripped the "Bearer " prefix, so `.credentials` is just the raw string
            status_code=401,
            detail={"error": {"code": "unauthorized", "message": "token is invalid"}},
        )
    return "alice"  # => the resolved identity -- unreachable from THIS example's own curl scenario,
    # => since the token sent below is deliberately wrong, but still needed for require_token to type-check
    # => (the type checker verifies EVERY code path returns str, reachable or not at runtime)
 
 
@app.get("/protected")  # => co-18: guarded entirely by the require_token dependency above --
# => the handler itself stays a single line, exactly like every open route -- FastAPI resolves
# => `require_token` BEFORE this function body runs, and never calls it at all if that dependency raises
def protected(user: str = Depends(require_token)) -> dict[str, str]:
    return {"user": user}  # => never actually reached when the curl below sends a garbage token

=== GET /protected (garbage token) ===
HTTP/1.1 401 Unauthorized
date: Tue, 14 Jul 2026 10:55:05 GMT
server: uvicorn
content-length: 62
content-type: application/json
 
{"error":{"code":"unauthorized","message":"token is invalid"}}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_malformed_token_is_401_with_invalid_message PASSED     [ 50%]
test_app.py::test_valid_token_still_reaches_handler_for_contrast 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 62: Valid Token -- a protected route with a good token succeeds."""
# => co-18: this is the THIRD scenario in the 401 trio (Examples 60, 61, 62) -- same dependency,
# => same guarded route, only the curl's Authorization header changes across all three examples
 
from fastapi import Depends, FastAPI, HTTPException, Request  # => co-23: Depends wires the auth check in
from fastapi.responses import JSONResponse  # => co-11: builds the exception handler's structured body
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer  # => co-18: parses "Bearer <token>"
 
app = FastAPI()  # => a fresh app -- this example needs no database, only auth plumbing
 
VALID_TOKEN = "s3cr3t-token-abc123"  # => hardcoded stand-in for a real signed/opaque token
security = HTTPBearer(auto_error=False)  # => auto_error=False: WE own the 401 body's shape below
 
 
@app.exception_handler(HTTPException)  # => co-11: unwraps FastAPI's default {"detail": ...} nesting
# => so a dict-shaped detail ships flat as {"error": {...}} -- same reusable pattern as ex-60/61
async def structured_error_handler(request: Request, exc: HTTPException) -> JSONResponse:
    body = (
        exc.detail
        if isinstance(exc.detail, dict)  # => the raise below always supplies a dict already
        else {"error": {"code": "error", "message": str(exc.detail)}}  # => fallback for a plain string
    )
    # => co-11: this handler is the ONLY place status codes get turned into response bodies --
    # => every route in this file stays completely free of response-formatting concerns
    return JSONResponse(status_code=exc.status_code, content=body)  # => co-11: same shape, every error
 
 
def require_token(  # => co-23: identical shape to ex-60/61 -- collapses the "missing" and "wrong"
    # => cases into ONE branch here, since this example's curl only ever needs to prove the OPPOSITE:
    # => that a genuinely correct token clears this check and reaches the handler body below
    credentials: HTTPAuthorizationCredentials | None = Depends(security),
) -> str:
    if credentials is None or credentials.credentials != VALID_TOKEN:  # => co-18: either failure mode
        raise HTTPException(
            status_code=401,
            detail={"error": {"code": "unauthorized", "message": "missing or invalid token"}},
        )  # => unreachable from THIS example's own curl -- kept only so the function is fully correct
        # => and equally usable if a reader copies this file and DOES send a bad token through it
    return "alice"  # => co-18: THIS example's focus -- the success path, reached only with a good token
    # => no session, no server-side lookup -- the token equality check above IS the entire proof
 
 
@app.get("/protected")  # => co-18: guarded entirely by the require_token dependency above
# => co-18: identical route SHAPE to ex-60/61 -- the only thing changing across the trio is the token
def protected(user: str = Depends(require_token)) -> dict[str, str | bool]:
    # => co-23: `user` is guaranteed to be "alice" here -- every failure path already returned 401
    return {"user": user, "granted": True}  # => a slightly richer body than ex-60/61 to mark success clearly

=== GET /protected (valid token) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:55:08 GMT
server: uvicorn
content-length: 31
content-type: application/json
 
{"user":"alice","granted":true}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_valid_token_returns_200_and_grants_access PASSED       [ 50%]
test_app.py::test_missing_and_invalid_still_fail_for_contrast 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 63: Protect Writes Only -- GET is open, POST/PUT/DELETE require a token."""
# => co-02, co-18: a very common real-world policy -- ANYONE can read the catalog, but only an
# => authenticated caller can change it. The HTTP method itself decides which rule applies below.
 
from fastapi import Depends, FastAPI, HTTPException, Request  # => co-02: method-sensitive protection
from fastapi.responses import JSONResponse  # => co-11: builds the exception handler's structured body
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer  # => co-18: parses "Bearer <token>"
 
app = FastAPI()  # => a fresh app -- this example needs no database, only an in-memory dict below
 
VALID_TOKEN = "s3cr3t-token-abc123"  # => hardcoded stand-in for a real signed/opaque token
security = HTTPBearer(auto_error=False)  # => auto_error=False: WE own the 401 body's shape below
ITEMS: dict[int, str] = {1: "milk", 2: "bread"}  # => in-memory store -- fine for this pedagogical example
# => (two seed rows so list_items() below has something visibly non-empty to return before any write)
NEXT_ID = 3  # => module-level counter for the next inserted id (co-05 caveat: not multi-worker safe --
# => Example 80 revisits this exact tension with a REAL, two-process-shared SQLite file instead)
 
 
@app.exception_handler(HTTPException)  # => co-11: one consistent {"error": {...}} envelope, unwrapped
async def structured_error_handler(request: Request, exc: HTTPException) -> JSONResponse:
    body = (
        exc.detail
        if isinstance(exc.detail, dict)  # => the raise below always supplies a dict already
        else {"error": {"code": "error", "message": str(exc.detail)}}  # => fallback for a plain string
    )
    return JSONResponse(status_code=exc.status_code, content=body)  # => co-11: same shape, every error
 
 
def require_token(  # => co-18: this dependency is opted into ONLY by write routes below --
    # => note the return type is None, not str -- this variant doesn't need the resolved identity,
    # => only the yes/no decision of whether the request is allowed to proceed at all
    credentials: HTTPAuthorizationCredentials | None = Depends(security),
) -> None:
    if credentials is None or credentials.credentials != VALID_TOKEN:  # => co-18: either failure mode
        raise HTTPException(
            status_code=401,
            detail={"error": {"code": "unauthorized", "message": "missing or invalid token"}},
        )
    # => co-18: no `return` needed -- reaching the end of this function IS the "allowed" outcome;
    # => FastAPI just runs the dependency for its side effect (raise-or-don't) and discards None
 
 
@app.get("/items")  # => co-02: GET has NO Depends(require_token) at all -- reads are open to everyone
# => co-02: read methods (GET, HEAD) are conventionally safe/idempotent -- open access here mirrors that
def list_items() -> dict[int, str]:
    return ITEMS  # => co-02: no auth check anywhere on this path -- curl with zero headers succeeds
 
 
@app.post("/items", dependencies=[Depends(require_token)])  # => co-02, co-18: WRITE guarded --
# => `dependencies=[...]` runs require_token for its SIDE EFFECT ONLY, without injecting a
# => parameter into create_item's own signature below -- the route itself never sees the token value
def create_item(name: str) -> dict[str, int | str]:
    global NEXT_ID  # => mutates the module-level counter -- co-05: this is process-local state,
    # => exactly the kind of in-memory mutation that breaks the moment a second worker exists
    item_id = NEXT_ID
    ITEMS[item_id] = name  # => the actual write this token protects
    NEXT_ID += 1  # => co-05: NOT atomic across concurrent requests in a real multi-threaded deployment
    return {"id": item_id, "name": name}  # => co-02: 200 by default here, unlike ex-19's explicit 201
 
 
@app.delete("/items/{item_id}", dependencies=[Depends(require_token)])  # => co-02, co-18: WRITE guarded --
# => same dependencies=[...] pattern as the POST route above -- consistent policy, one line each
# => co-02: DELETE, POST, and (in a real app) PUT/PATCH all share this exact same guard style
def delete_item(item_id: int) -> dict[str, bool]:
    ITEMS.pop(item_id, None)  # => the actual write this token protects -- second-arg None means
    # => "don't raise KeyError if item_id doesn't exist," making delete naturally idempotent (co-02)
    return {"deleted": True}  # => co-02: reports success even if the id was already gone -- by design

=== GET /items (open, no token) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:55:10 GMT
server: uvicorn
content-length: 24
content-type: application/json
 
{"1":"milk","2":"bread"}
=== POST /items?name=eggs (no token, 401) ===
HTTP/1.1 401 Unauthorized
date: Tue, 14 Jul 2026 10:55:10 GMT
server: uvicorn
content-length: 70
content-type: application/json
 
{"error":{"code":"unauthorized","message":"missing or invalid token"}}
=== POST /items?name=eggs (valid token) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:55:10 GMT
server: uvicorn
content-length: 22
content-type: application/json
 
{"id":3,"name":"eggs"}
=== GET /items (after write) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:55:10 GMT
server: uvicorn
content-length: 35
content-type: application/json
 
{"1":"milk","2":"bread","3":"eggs"}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 4 items
 
test_app.py::test_get_is_open_without_any_token PASSED                   [ 25%]
test_app.py::test_post_without_token_is_401 PASSED                       [ 50%]
test_app.py::test_post_with_token_succeeds PASSED                        [ 75%]
test_app.py::test_delete_without_token_is_401_but_get_still_open 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
========================= 4 passed, 1 warning in 0.19s =========================

"""Example 64: Session-Cookie Auth -- set a session cookie on login, read it next request."""
# => co-17: this is Example 57's session mechanism made into its OWN complete, runnable app --
# => login mints server-side state, the browser only ever holds an opaque cookie pointing at it
 
import uuid  # => generates the unguessable session id -- never a sequential/predictable value
 
from fastapi import Cookie, FastAPI, HTTPException, Request, Response  # => co-17: cookie-based session
from fastapi.responses import JSONResponse  # => co-11: builds the exception handler's structured body
 
app = FastAPI()  # => a fresh app -- this example needs no database, only the in-memory dict below
# => (fully self-contained: nothing here is imported from any other example directory)
 
SESSIONS: dict[str, str] = {}  # => session_id -> username; server-side state (co-17) -- this dict
# => IS the session store; a real deployment would put this in Redis/a DB, precisely because
# => in-process memory vanishes on restart and isn't shared across multiple worker processes
USERNAME = "alice"  # => the ONE registered user this pedagogical example recognizes
PASSWORD = "wonderland"  # => a hardcoded password -- never do this in real code (co-17 caveat)
# => (a real login endpoint hashes and salts stored passwords -- this constant exists purely
# => so curl has something deterministic to submit; nothing here is meant as production auth advice)
 
 
@app.exception_handler(HTTPException)  # => co-11: consistent envelope, same pattern as ex-60..63
async def structured_error_handler(request: Request, exc: HTTPException) -> JSONResponse:
    body = (
        exc.detail
        if isinstance(exc.detail, dict)  # => every raise below supplies a dict already
        else {"error": {"code": "error", "message": str(exc.detail)}}  # => fallback for a plain string
    )
    return JSONResponse(status_code=exc.status_code, content=body)  # => co-11: same shape, every error
 
 
@app.post("/login")  # => co-17: authenticates, then ESTABLISHES a session -- unlike Example 58's
# => token endpoint, this one has a SIDE EFFECT: it writes a new entry into SESSIONS below
def login(username: str, password: str, response: Response) -> dict[str, str]:
    if username != USERNAME or password != PASSWORD:  # => co-17: reject any mismatch immediately --
        # => deliberately the SAME message for both a wrong username and a wrong password --
        # => never let a caller distinguish "no such user" from "wrong password" via the response
        raise HTTPException(
            status_code=401,
            detail={"error": {"code": "unauthorized", "message": "invalid credentials"}},
        )
    session_id = str(uuid.uuid4())  # => a fresh, unguessable session identifier -- a real deployment
    # => also rotates this on every login, so an old leaked cookie can't be replayed indefinitely
    SESSIONS[session_id] = username  # => co-17: the SERVER remembers this session belongs to username
    response.set_cookie(  # => co-04: writes a Set-Cookie response header -- curl's -c flag captures it
        "session_id", session_id, httponly=True
    )  # => co-04: httponly prevents client-side JS from reading it -- mitigates cookie-theft via XSS
    return {"logged_in_as": username}  # => co-17: confirms WHO logged in -- never the session id itself
    # => (leaking the session id in the JSON body too would defeat httponly's whole protection above)
 
 
@app.get("/me")  # => co-17: identifies the caller purely from the session cookie on THIS request --
# => curl's -b flag is what actually SENDS the cookie captured by -c above back to this endpoint
def me(session_id: str | None = Cookie(default=None)) -> dict[str, str]:
    if session_id is None or session_id not in SESSIONS:  # => no cookie, or the server never issued it
        # => (also covers a server restart, since SESSIONS is in-memory and empties on every reload)
        raise HTTPException(
            status_code=401,
            detail={"error": {"code": "unauthorized", "message": "no active session"}},
        )
    return {"username": SESSIONS[session_id]}  # => the SAME identity established at login, now confirmed
    # => co-17: notice this handler never re-checks a password -- the cookie's PRESENCE in
    # => SESSIONS is the entire proof of identity for every request after the original login

=== POST /login (writes cookie jar) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:55:13 GMT
server: uvicorn
content-length: 24
content-type: application/json
set-cookie: session_id=b781451b-6fb9-492d-b828-a25f2d7c3ea1; HttpOnly; Path=/; SameSite=lax
 
{"logged_in_as":"alice"}
=== cookie jar contents ===
# Netscape HTTP Cookie File
# https://curl.se/docs/http-cookies.html
# This file was generated by libcurl! Edit at your own risk.
 
#HttpOnly_localhost    FALSE    /    FALSE    0    session_id    b781451b-6fb9-492d-b828-a25f2d7c3ea1
=== GET /me (reads cookie jar) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:55:13 GMT
server: uvicorn
content-length: 20
content-type: application/json
 
{"username":"alice"}
=== GET /me (no cookie jar) ===
HTTP/1.1 401 Unauthorized
date: Tue, 14 Jul 2026 10:55:13 GMT
server: uvicorn
content-length: 63
content-type: application/json
 
{"error":{"code":"unauthorized","message":"no active session"}}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_login_sets_cookie_and_me_reads_it_back PASSED          [ 33%]
test_app.py::test_me_without_prior_login_is_401 PASSED                   [ 66%]
test_app.py::test_login_with_wrong_password_never_sets_a_cookie 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 65: Pagination limit/offset -- GET /tasks?limit=&offset= slices the list."""
 
import os
import sqlite3
from typing import TypedDict
 
from fastapi import FastAPI, Query  # => co-12: limit/offset are typed QUERY parameters
 
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")  # => co-14: a real on-disk SQLite file,
# => one PER EXAMPLE directory -- never shared with any other example, keeping each self-contained
 
# => co-15: the schema below backs every list query in this example. Column-by-column:
# =>   id          -- an auto-incrementing primary key, never chosen by the caller
# =>   title       -- the task's display text, no constraint on it beyond NOT NULL
# =>   status      -- co-20: a FILTERABLE field, exercised starting at Example 69
# =>   priority    -- co-20: a SECOND filterable field, exercised starting at Example 70
# =>   created_at  -- co-20: a SORTABLE field, exercised starting at Example 72
SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    status TEXT NOT NULL,
    priority TEXT NOT NULL,
    created_at TEXT NOT NULL
);
"""
 
 
class TaskRow(TypedDict):  # => co-14, co-24: the repository's typed return shape
    id: int  # => matches the schema's INTEGER PRIMARY KEY column
    title: str  # => matches the schema's title column
    status: str  # => matches the schema's status column
    priority: str  # => matches the schema's priority column
    created_at: str  # => matches the schema's created_at column
 
 
def _seed(conn: sqlite3.Connection) -> None:
    statuses = ["todo", "in_progress", "done"]  # => rotates every 3 rows -- index is i % 3
    priorities = ["low", "normal", "high"]  # => rotates every 2 rows via (i // 2) % 3 -- DECORRELATED
    # => from status ON PURPOSE, so that filtering by BOTH fields together (Example 70) genuinely
    # => narrows the result further than either single filter alone, instead of the two columns
    # => always moving in lockstep (which would make a combined filter demonstrate nothing new)
    rows = [
        (f"task {i}", statuses[i % 3], priorities[(i // 2) % 3], f"2026-07-{i:02d}T00:00:00")
        for i in range(1, 26)  # => 25 rows -- enough to make a small page window visibly DIFFERENT
        # => from the full list, unlike a 1-2 row toy dataset where pagination wouldn't be provable
    ]
    conn.executemany(  # => co-14: a single batched INSERT for all 25 rows, one transaction
        "INSERT INTO tasks (title, status, priority, created_at) VALUES (?, ?, ?, ?)",  # => co-14: ?
        # => placeholders -- the SAME parameterization principle Example 71 stress-tests directly
        rows,
    )
    conn.commit()  # => co-14: commits the whole seed batch at once
 
 
def init_db() -> None:  # => co-15: fresh schema + seed data every time this module is imported
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)  # => start from a known, deterministic state -- this example's expected
        # => row counts and ids only hold true starting from a CLEAN file, never an accumulated one
    conn = sqlite3.connect(DB_PATH)
    conn.execute(SCHEMA)  # => co-15: creates the table every query below depends on existing
    _seed(conn)  # => populates it with the 25-row deterministic dataset described above
    conn.close()  # => co-14: connections are short-lived here -- opened, used, closed, never held
 
 
def get_connection() -> sqlite3.Connection:  # => co-14: the repository's ONLY entry point to the DB --
    # => every query function below calls THIS instead of sqlite3.connect() directly
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # => rows behave like dicts -- readable by column name, which is
    # => what makes `dict(row)` below produce sensible {"id":..., "title":...} keys, not just a tuple
    return conn  # => a fresh connection per call -- co-14: no pooling, matches this example's scale
 
 
def list_tasks(limit: int, offset: int) -> list[TaskRow]:  # => co-14, co-19: the repository function itself
    # => owns 100% of the SQL for this example -- co-24: the ONLY place LIMIT/OFFSET semantics live
    conn = get_connection()
    cursor = conn.execute(  # => co-14: LIMIT/OFFSET are parameterized, never string-interpolated --
        # => an f-string here would reopen the exact SQL-injection risk Example 71 stress-tests
        "SELECT id, title, status, priority, created_at FROM tasks ORDER BY id LIMIT ? OFFSET ?",
        (limit, offset),  # => co-19: THIS example's core mechanism -- limit caps the page size,
        # => offset skips ahead -- the exact two numbers a caller controls via query params below
    )  # => co-15: ORDER BY id keeps page boundaries STABLE across calls -- without an explicit
    # => order, SQLite offers no guarantee that repeated queries return rows in the same sequence
    result = [dict(row) for row in cursor.fetchall()]  # type: ignore[misc]  # => sqlite3.Row -> plain dict
    conn.close()  # => co-14: closed immediately after this one query -- no connection pooling needed here
    return result  # type: ignore[return-value]  # => shape matches TaskRow at runtime
 
 
init_db()  # => co-15: runs once at import time, before the app starts serving -- every curl call
# => below sees the SAME 25-row dataset, since nothing in this example ever re-seeds mid-run
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the pagination mechanism
 
 
@app.get("/tasks")  # => co-19: the paginated list endpoint this whole example is about --
# => FastAPI parses `?limit=&offset=` from the URL and hands them to this function as plain ints
def get_tasks(
    limit: int = Query(default=10, ge=1),  # => co-12: a typed, constrained query parameter --
    # => ge=1 rejects zero/negative limits before the handler body ever runs (co-10) --
    # => a caller who omits `?limit=` entirely gets the default of 10, proven by Example 66
    offset: int = Query(default=0, ge=0),  # => co-12: how many rows to skip before the page starts --
    # => ge=0 rejects a negative offset the same way limit rejects a non-positive value above
) -> list[TaskRow]:
    return list_tasks(limit, offset)  # => co-19: the handler holds NO SQL -- co-24 layering --
    # => every subsequent pagination/filter/sort example in this topic reuses this exact split:
    # => a thin route function delegating to a repository function that owns the actual query

=== GET /tasks (default limit=10, offset=0) ===
[{"id":1,"title":"task 1","status":"in_progress","priority":"low","created_at":"2026-07-01T00:00:00"},{"id":2,"title":"task 2","status":"done","priority":"normal","created_at":"2026-07-02T00:00:00"},{"id":3,"title":"task 3","status":"todo","priority":"normal","created_at":"2026-07-03T00:00:00"},{"id":4,"title":"task 4","status":"in_progress","priority":"high","created_at":"2026-07-04T00:00:00"},{"id":5,"title":"task 5","status":"done","priority":"high","created_at":"2026-07-05T00:00:00"},{"id":6,"title":"task 6","status":"todo","priority":"low","created_at":"2026-07-06T00:00:00"},{"id":7,"title":"task 7","status":"in_progress","priority":"low","created_at":"2026-07-07T00:00:00"},{"id":8,"title":"task 8","status":"done","priority":"normal","created_at":"2026-07-08T00:00:00"},{"id":9,"title":"task 9","status":"todo","priority":"normal","created_at":"2026-07-09T00:00:00"},{"id":10,"title":"task 10","status":"in_progress","priority":"high","created_at":"2026-07-10T00:00:00"}]
=== GET /tasks?limit=5&offset=10 ===
[{"id":11,"title":"task 11","status":"done","priority":"high","created_at":"2026-07-11T00:00:00"},{"id":12,"title":"task 12","status":"todo","priority":"low","created_at":"2026-07-12T00:00:00"},{"id":13,"title":"task 13","status":"in_progress","priority":"low","created_at":"2026-07-13T00:00:00"},{"id":14,"title":"task 14","status":"done","priority":"normal","created_at":"2026-07-14T00:00:00"},{"id":15,"title":"task 15","status":"todo","priority":"normal","created_at":"2026-07-15T00:00:00"}]

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_default_page_returns_first_ten PASSED                  [ 33%]
test_app.py::test_limit_5_offset_10_returns_a_different_window PASSED    [ 66%]
test_app.py::test_offset_past_the_end_returns_empty 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 66: Pagination Default -- a default limit applies when the param is absent."""
# => co-12, co-19: builds on the SAME limit/offset mechanism as Example 65, this time proving
# => what happens on the OTHER side -- a caller who sends no limit at all still gets a bounded page
 
import os
import sqlite3  # => co-14: the stdlib DB driver -- no ORM, no extra dependency needed
from typing import TypedDict  # => co-09: a typed dict shape for what the repository returns
 
from fastapi import FastAPI, Query  # => co-12: query params are typed and validated here
# => (fully self-contained: nothing here is imported from any other example directory)
 
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")  # => co-14: one on-disk SQLite
# => file PER EXAMPLE DIRECTORY -- never shared with any other example, keeping each self-contained
 
# => co-15: the schema below backs every list query in this example. Column-by-column:
# =>   id          -- an auto-incrementing primary key, never chosen by the caller
# =>   title       -- the task's display text, no constraint on it beyond NOT NULL
# =>   status      -- co-20: a FILTERABLE field, exercised starting at Example 69
# =>   priority    -- co-20: a SECOND filterable field, exercised starting at Example 70
# =>   created_at  -- co-20: a SORTABLE field, exercised starting at Example 72
SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    status TEXT NOT NULL,
    priority TEXT NOT NULL,
    created_at TEXT NOT NULL
);
"""
 
 
class TaskRow(TypedDict):  # => co-14, co-24: the repository's typed return shape --
    # => every field below matches a SCHEMA column one-to-one, so `dict(row)` (further down)
    # => produces a value that satisfies this shape without any manual field-by-field mapping
    id: int  # => matches the schema's INTEGER PRIMARY KEY column
    title: str  # => matches the schema's title column
    status: str  # => matches the schema's status column
    priority: str  # => matches the schema's priority column
    created_at: str  # => matches the schema's created_at column
 
 
def _seed(conn: sqlite3.Connection) -> None:  # => co-14: builds the SAME deterministic 25-row
    # => dataset used by every pagination/filter/sort example in this topic, so worked-out ids
    # => like "done={2,5,8,...}" stay correct however many of these example directories exist
    statuses = ["todo", "in_progress", "done"]  # => rotates every 3 rows -- index is i % 3
    priorities = ["low", "normal", "high"]  # => rotates every 2 rows via (i // 2) % 3 --
    # => DECORRELATED from status ON PURPOSE, so a combined status+priority filter genuinely
    # => narrows further than either field alone, instead of the two columns moving in lockstep
    rows = [
        (f"task {i}", statuses[i % 3], priorities[(i // 2) % 3], f"2026-07-{i:02d}T00:00:00")
        for i in range(1, 26)  # => the SAME 25-row seed used across the pagination/filter examples --
        # => 25 is more than DEFAULT_LIMIT, so the default page is PROVABLY shorter than the full set
    ]
    conn.executemany(  # => co-14: a single batched INSERT for all 25 rows, one transaction
        "INSERT INTO tasks (title, status, priority, created_at) VALUES (?, ?, ?, ?)",  # => co-14:
        # => ? placeholders -- the SAME parameterization principle Example 71 stress-tests directly
        rows,
    )
    conn.commit()  # => co-14: commits the whole seed batch at once, not row-by-row
 
 
def init_db() -> None:  # => co-15: fresh schema + seed data every time this module is imported
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)  # => start from a known, deterministic state -- this example's expected
        # => row counts and ids only hold true starting from a CLEAN file, never an accumulated one
    conn = sqlite3.connect(DB_PATH)
    conn.execute(SCHEMA)  # => co-15: creates the table every query below depends on existing
    _seed(conn)  # => populates it with the 25-row deterministic dataset described above
    conn.close()  # => co-14: connections are short-lived here -- opened, used, closed, never held
 
 
def get_connection() -> sqlite3.Connection:  # => co-14: the repository's ONLY entry point to
    # => the DB -- every query function below calls THIS instead of sqlite3.connect() directly
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # => rows behave like dicts -- readable by column name, which
    # => is what makes `dict(row)` below produce sensible {"id":..., "title":...} keys, not a tuple
    return conn  # => a fresh connection per call -- co-14: no pooling, matches this example's scale
 
 
DEFAULT_LIMIT = 10  # => co-19: a NAMED constant -- the exact bound applied when the caller sends
# => nothing at all for `?limit=` -- naming it (rather than a bare 10 inline) means the value is
# => documented once, reused everywhere below, and trivial to find without reading every route
 
 
def list_tasks(limit: int, offset: int) -> list[TaskRow]:  # => co-14, co-19: identical query shape
    # => to every other example in this cluster -- what's NEW here is what CALLS it, further down
    conn = get_connection()
    cursor = conn.execute(  # => co-14: LIMIT/OFFSET are parameterized, never string-interpolated
        "SELECT id, title, status, priority, created_at FROM tasks ORDER BY id LIMIT ? OFFSET ?",
        (limit, offset),  # => co-19: whatever the caller sent (or the default below) lands here
    )
    result = [dict(row) for row in cursor.fetchall()]  # type: ignore[misc]  # => sqlite3.Row -> dict
    conn.close()  # => co-14: closed immediately after this one query
    return result  # type: ignore[return-value]  # => shape matches TaskRow at runtime
 
 
def count_tasks() -> int:  # => co-19: total row count, used here only to PROVE the default is bounded --
    # => this example's curl calls THIS endpoint to show 25 rows exist, then /tasks to show only 10 return
    conn = get_connection()
    total = conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0]  # => co-14: a single scalar query
    conn.close()
    return int(total)  # => SQLite returns a plain int already -- the cast documents the intent, not a fix
 
 
init_db()  # => co-15: runs once at import time, before the app starts serving
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the default-limit mechanism
 
 
@app.get("/tasks")  # => co-19: this example's focus -- what happens when limit is OMITTED entirely
def get_tasks(
    limit: int = Query(default=DEFAULT_LIMIT, ge=1),  # => co-12: the default only kicks in when
    # => `?limit=` is absent from the URL altogether -- sending `?limit=0` would instead hit ge=1's
    # => validation error, which is a DIFFERENT scenario than "the caller said nothing" (this example's)
    offset: int = Query(default=0, ge=0),  # => co-12: same default-when-absent behavior, for offset
) -> list[TaskRow]:
    return list_tasks(limit, offset)  # => co-19: the handler holds NO SQL -- co-24 layering
 
 
@app.get("/tasks/count")  # => a small helper endpoint proving 25 rows genuinely exist behind the default
def get_count() -> dict[str, int]:
    return {"total": count_tasks()}  # => co-19: contrast THIS number against /tasks's page length

=== GET /tasks/count (proves 25 rows exist) ===
{"total":25}
=== GET /tasks (no limit param at all -- default applies) ===
row count: 10
ids: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_omitting_limit_uses_the_default_not_the_full_dataset PASSED [ 50%]
test_app.py::test_explicit_limit_still_overrides_the_default 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 67: Pagination Metadata -- the response includes total and next."""
# => co-19: Example 65 proved limit/offset SLICES the list; this example proves a client can also
# => discover how many pages exist and where the next one starts, without a separate count request
 
import os
import sqlite3  # => co-14: the stdlib DB driver -- no ORM, no extra dependency needed
from typing import TypedDict  # => co-09: a typed dict shape for what the repository returns
 
from fastapi import FastAPI, Query  # => co-12: query params are typed and validated here
# => (fully self-contained: nothing here is imported from any other example directory)
 
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")  # => co-14: one on-disk SQLite
# => file PER EXAMPLE DIRECTORY -- never shared with any other example, keeping each self-contained
 
# => co-15: the schema below backs every list query in this example. Column-by-column:
# =>   id          -- an auto-incrementing primary key, never chosen by the caller
# =>   title       -- the task's display text, no constraint on it beyond NOT NULL
# =>   status      -- co-20: a FILTERABLE field, exercised starting at Example 69
# =>   priority    -- co-20: a SECOND filterable field, exercised starting at Example 70
# =>   created_at  -- co-20: a SORTABLE field, exercised starting at Example 72
SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    status TEXT NOT NULL,
    priority TEXT NOT NULL,
    created_at TEXT NOT NULL
);
"""
 
 
class TaskRow(TypedDict):  # => co-14, co-24: the repository's typed return shape --
    # => every field below matches a SCHEMA column one-to-one, so `dict(row)` (further down)
    # => produces a value that satisfies this shape without any manual field-by-field mapping
    id: int  # => matches the schema's INTEGER PRIMARY KEY column
    title: str  # => matches the schema's title column
    status: str  # => matches the schema's status column
    priority: str  # => matches the schema's priority column
    created_at: str  # => matches the schema's created_at column
 
 
class Page(TypedDict):  # => co-09, co-19: the ENVELOPE shape -- not just a bare array anymore --
    # => every list endpoint from THIS example onward returns this shape instead of a raw list
    items: list[TaskRow]  # => this page's rows -- exactly what Example 65 returned on its own
    total: int  # => co-19: the FULL table's row count, independent of this page's size
    next: int | None  # => None means "no further page" -- a real, typed sentinel, not a magic number
 
 
def _seed(conn: sqlite3.Connection) -> None:  # => co-14: builds the SAME deterministic 25-row
    # => dataset used by every pagination/filter/sort example in this topic, so worked-out ids
    # => like "done={2,5,8,...}" stay correct however many of these example directories exist
    statuses = ["todo", "in_progress", "done"]  # => rotates every 3 rows -- index is i % 3
    priorities = ["low", "normal", "high"]  # => co-19: irrelevant to pagination itself, but kept
    # => identical to Example 65's schema/seed so this file stays a complete, self-contained app --
    # => rotates every 2 rows via (i // 2) % 3, DECORRELATED from status ON PURPOSE (a combined
    # => status+priority filter, exercised starting at Example 70, needs the two fields independent)
    rows = [
        (f"task {i}", statuses[i % 3], priorities[(i // 2) % 3], f"2026-07-{i:02d}T00:00:00")
        for i in range(1, 26)  # => 25 total rows -- big enough that a 10-row page genuinely has a next
    ]
    conn.executemany(  # => co-14: a single batched INSERT for all 25 rows, one transaction
        "INSERT INTO tasks (title, status, priority, created_at) VALUES (?, ?, ?, ?)",  # => co-14:
        # => ? placeholders -- the SAME parameterization principle Example 71 stress-tests directly
        rows,
    )
    conn.commit()  # => co-14: commits the whole seed batch at once, not row-by-row
 
 
def init_db() -> None:  # => co-15: fresh schema + seed data every time this module is imported
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)  # => start from a known, deterministic state -- this example's expected
        # => row counts and ids only hold true starting from a CLEAN file, never an accumulated one
    conn = sqlite3.connect(DB_PATH)
    conn.execute(SCHEMA)  # => co-15: creates the table every query below depends on existing
    _seed(conn)  # => populates it with the 25-row deterministic dataset described above
    conn.close()  # => co-14: connections are short-lived here -- opened, used, closed, never held
 
 
def get_connection() -> sqlite3.Connection:  # => co-14: the repository's ONLY entry point to
    # => the DB -- every query function below calls THIS instead of sqlite3.connect() directly
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # => rows behave like dicts -- readable by column name, which
    # => is what makes `dict(row)` below produce sensible {"id":..., "title":...} keys, not a tuple
    return conn  # => a fresh connection per call -- co-14: no pooling, matches this example's scale
 
 
def list_page(limit: int, offset: int) -> Page:  # => co-14, co-19: repository returns the FULL
    # => envelope this time -- not just the rows, but ENOUGH metadata for a client to keep paging
    conn = get_connection()
    total = int(conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0])  # => co-19: the WHOLE
    # => table's size, computed with a SEPARATE query -- the LIMIT below never touches this number
    cursor = conn.execute(  # => co-14: the exact same paginated SELECT as Example 65
        "SELECT id, title, status, priority, created_at FROM tasks ORDER BY id LIMIT ? OFFSET ?",
        (limit, offset),
    )
    items = [dict(row) for row in cursor.fetchall()]  # type: ignore[misc]  # => sqlite3.Row -> dict
    conn.close()  # => co-14: closed immediately after both queries above finish
    next_offset = offset + limit  # => co-19: the offset a client would use to fetch the FOLLOWING page
    has_more = next_offset < total  # => whether that following page would return anything at all --
    # => an offset landing exactly ON or past `total` means there is nothing left to return
    return {
        "items": items,  # type: ignore[typeddict-item]  # => this page's slice of rows
        "total": total,  # => co-19: the FULL count, letting a client compute "page 3 of N" itself
        "next": next_offset if has_more else None,  # => None signals "you have reached the end" --
        # => a typed sentinel a client can check with `if page["next"] is not None`, no magic number
    }
 
 
init_db()  # => co-15: runs once at import time, before the app starts serving
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the pagination envelope
 
 
@app.get("/tasks")  # => co-19, co-09: returns the metadata envelope, not a bare list -- Example 65's
# => `list[TaskRow]` return type becomes `Page` here, wrapping the same rows with total/next alongside
def get_tasks(
    limit: int = Query(default=10, ge=1),  # => co-12: identical constraint to Example 65
    offset: int = Query(default=0, ge=0),  # => co-12: identical constraint to Example 65
) -> Page:
    return list_page(limit, offset)  # => co-19: the handler holds NO SQL -- co-24 layering

=== GET /tasks?limit=10&offset=0 (first page) ===
{"items":[{"id":1,"title":"task 1","status":"in_progress","priority":"low","created_at":"2026-07-01T00:00:00"},{"id":2,"title":"task 2","status":"done","priority":"normal","created_at":"2026-07-02T00:00:00"},{"id":3,"title":"task 3","status":"todo","priority":"normal","created_at":"2026-07-03T00:00:00"},{"id":4,"title":"task 4","status":"in_progress","priority":"high","created_at":"2026-07-04T00:00:00"},{"id":5,"title":"task 5","status":"done","priority":"high","created_at":"2026-07-05T00:00:00"},{"id":6,"title":"task 6","status":"todo","priority":"low","created_at":"2026-07-06T00:00:00"},{"id":7,"title":"task 7","status":"in_progress","priority":"low","created_at":"2026-07-07T00:00:00"},{"id":8,"title":"task 8","status":"done","priority":"normal","created_at":"2026-07-08T00:00:00"},{"id":9,"title":"task 9","status":"todo","priority":"normal","created_at":"2026-07-09T00:00:00"},{"id":10,"title":"task 10","status":"in_progress","priority":"high","created_at":"2026-07-10T00:00:00"}],"total":25,"next":10}
=== GET /tasks?limit=10&offset=20 (last page) ===
{"items":[{"id":21,"title":"task 21","status":"todo","priority":"normal","created_at":"2026-07-21T00:00:00"},{"id":22,"title":"task 22","status":"in_progress","priority":"high","created_at":"2026-07-22T00:00:00"},{"id":23,"title":"task 23","status":"done","priority":"high","created_at":"2026-07-23T00:00:00"},{"id":24,"title":"task 24","status":"todo","priority":"low","created_at":"2026-07-24T00:00:00"},{"id":25,"title":"task 25","status":"in_progress","priority":"low","created_at":"2026-07-25T00:00:00"}],"total":25,"next":null}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_first_page_reports_total_and_a_next_offset PASSED      [ 50%]
test_app.py::test_last_page_reports_next_as_null 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 68: Pagination Bounds -- a limit over the maximum is rejected with 422."""
# => co-10, co-19: Example 65's `ge=1` stopped a limit too SMALL; this example adds `le=MAX_LIMIT`
# => to stop one too LARGE -- together they turn an unbounded page size into a safely bounded one
 
import os
import sqlite3  # => co-14: the stdlib DB driver -- no ORM, no extra dependency needed
from typing import TypedDict  # => co-09: a typed dict shape for what the repository returns
 
from fastapi import FastAPI, Query  # => co-12: query params are typed and validated here
# => (fully self-contained: nothing here is imported from any other example directory)
 
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")  # => co-14: one on-disk SQLite
# => file PER EXAMPLE DIRECTORY -- never shared with any other example, keeping each self-contained
 
# => co-15: the schema below backs every list query in this example. Column-by-column:
# =>   id          -- an auto-incrementing primary key, never chosen by the caller
# =>   title       -- the task's display text, no constraint on it beyond NOT NULL
# =>   status      -- co-20: a FILTERABLE field, exercised starting at Example 69
# =>   priority    -- co-20: a SECOND filterable field, exercised starting at Example 70
# =>   created_at  -- co-20: a SORTABLE field, exercised starting at Example 72
SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    status TEXT NOT NULL,
    priority TEXT NOT NULL,
    created_at TEXT NOT NULL
);
"""
 
 
class TaskRow(TypedDict):  # => co-14, co-24: the repository's typed return shape --
    # => every field below matches a SCHEMA column one-to-one, so `dict(row)` (further down)
    # => produces a value that satisfies this shape without any manual field-by-field mapping
    id: int  # => matches the schema's INTEGER PRIMARY KEY column
    title: str  # => matches the schema's title column
    status: str  # => matches the schema's status column
    priority: str  # => matches the schema's priority column
    created_at: str  # => matches the schema's created_at column
 
 
def _seed(conn: sqlite3.Connection) -> None:  # => co-14: builds the SAME deterministic 25-row
    # => dataset used by every pagination/filter/sort example in this topic, so worked-out ids
    # => like "done={2,5,8,...}" stay correct however many of these example directories exist
    statuses = ["todo", "in_progress", "done"]  # => rotates every 3 rows -- index is i % 3
    priorities = ["low", "normal", "high"]  # => rotates every 2 rows via (i // 2) % 3 --
    # => DECORRELATED from status ON PURPOSE, so a combined status+priority filter genuinely
    # => narrows further than either field alone, instead of the two columns moving in lockstep
    rows = [
        (f"task {i}", statuses[i % 3], priorities[(i // 2) % 3], f"2026-07-{i:02d}T00:00:00")
        for i in range(1, 26)  # => 25 rows -- fewer than MAX_LIMIT, so a request AT the max still succeeds
    ]
    conn.executemany(  # => co-14: a single batched INSERT for all 25 rows, one transaction
        "INSERT INTO tasks (title, status, priority, created_at) VALUES (?, ?, ?, ?)",  # => co-14:
        # => ? placeholders -- the SAME parameterization principle Example 71 stress-tests directly
        rows,
    )
    conn.commit()  # => co-14: commits the whole seed batch at once, not row-by-row
 
 
def init_db() -> None:  # => co-15: fresh schema + seed data every time this module is imported
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)  # => start from a known, deterministic state -- this example's expected
        # => row counts and ids only hold true starting from a CLEAN file, never an accumulated one
    conn = sqlite3.connect(DB_PATH)
    conn.execute(SCHEMA)  # => co-15: creates the table every query below depends on existing
    _seed(conn)  # => populates it with the 25-row deterministic dataset described above
    conn.close()  # => co-14: connections are short-lived here -- opened, used, closed, never held
 
 
def get_connection() -> sqlite3.Connection:  # => co-14: the repository's ONLY entry point to
    # => the DB -- every query function below calls THIS instead of sqlite3.connect() directly
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # => rows behave like dicts -- readable by column name, which
    # => is what makes `dict(row)` below produce sensible {"id":..., "title":...} keys, not a tuple
    return conn  # => a fresh connection per call -- co-14: no pooling, matches this example's scale
 
 
MAX_LIMIT = 50  # => co-19: the ceiling this example enforces -- prevents an accidental "select
# => everything" call from a caller who passes an enormous limit like ?limit=1000000 on a huge table
 
 
def list_tasks(limit: int, offset: int) -> list[TaskRow]:  # => co-14, co-19: identical query shape
    # => to Example 65 -- this function TRUSTS `limit` is already within bounds by the time it runs
    conn = get_connection()
    cursor = conn.execute(  # => co-14: LIMIT/OFFSET are parameterized, never string-interpolated
        "SELECT id, title, status, priority, created_at FROM tasks ORDER BY id LIMIT ? OFFSET ?",
        (limit, offset),
    )
    result = [dict(row) for row in cursor.fetchall()]  # type: ignore[misc]  # => sqlite3.Row -> dict
    conn.close()  # => co-14: closed immediately after this one query
    return result  # type: ignore[return-value]  # => shape matches TaskRow at runtime
 
 
init_db()  # => co-15: runs once at import time, before the app starts serving
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the bounds-checking mechanism
 
 
@app.get("/tasks")  # => co-19, co-10: the guard is declared right in the query parameter's constraints --
# => no `if limit > 50: raise ...` anywhere in this file; FastAPI enforces the bound declaratively
def get_tasks(
    limit: int = Query(default=10, ge=1, le=MAX_LIMIT),  # => co-10: le=50 -- FastAPI REJECTS anything
    # => higher with a 422 before this function is even entered, same enforcement point as ge=1 below
    offset: int = Query(default=0, ge=0),  # => co-12: unrelated to THIS example's bound, kept for parity
) -> list[TaskRow]:
    return list_tasks(limit, offset)  # => this line never runs at all for an out-of-bounds limit -- co-10
    # => validation happens BEFORE the handler body, exactly like any other Pydantic/Query constraint

=== GET /tasks?limit=50 (at the max -- succeeds) ===
row count: 25
=== GET /tasks?limit=1000 (over the max -- 422) ===
HTTP/1.1 422 Unprocessable Content
date: Tue, 14 Jul 2026 10:55:47 GMT
server: uvicorn
content-length: 143
content-type: application/json
 
{"detail":[{"type":"less_than_equal","loc":["query","limit"],"msg":"Input should be less than or equal to 50","input":"1000","ctx":{"le":50}}]}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_limit_over_maximum_is_422 PASSED                       [ 33%]
test_app.py::test_limit_at_maximum_is_allowed PASSED                     [ 66%]
test_app.py::test_limit_zero_is_422 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 69: Filter by Field -- ?status=done narrows the list."""
# => co-20: builds on Example 65's list endpoint by adding a NARROWING dimension alongside
# => pagination -- a caller now controls WHICH rows qualify, not just how many come back per page
 
import os
import sqlite3  # => co-14: the stdlib DB driver -- no ORM, no extra dependency needed
from typing import TypedDict  # => co-09: a typed dict shape for what the repository returns
 
from fastapi import FastAPI, Query  # => co-12: query params are typed and validated here
# => (fully self-contained: nothing here is imported from any other example directory)
 
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")  # => co-14: one on-disk SQLite
# => file PER EXAMPLE DIRECTORY -- never shared with any other example, keeping each self-contained
 
# => co-15: the schema below backs every list query in this example. Column-by-column:
# =>   id          -- an auto-incrementing primary key, never chosen by the caller
# =>   title       -- the task's display text, no constraint on it beyond NOT NULL
# =>   status      -- co-20: a FILTERABLE field, exercised starting at Example 69
# =>   priority    -- co-20: a SECOND filterable field, exercised starting at Example 70
# =>   created_at  -- co-20: a SORTABLE field, exercised starting at Example 72
SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    status TEXT NOT NULL,
    priority TEXT NOT NULL,
    created_at TEXT NOT NULL
);
"""
 
 
class TaskRow(TypedDict):  # => co-14, co-24: the repository's typed return shape --
    # => every field below matches a SCHEMA column one-to-one, so `dict(row)` (further down)
    # => produces a value that satisfies this shape without any manual field-by-field mapping
    id: int  # => matches the schema's INTEGER PRIMARY KEY column
    title: str  # => matches the schema's title column
    status: str  # => matches the schema's status column
    priority: str  # => matches the schema's priority column
    created_at: str  # => matches the schema's created_at column
 
 
def _seed(conn: sqlite3.Connection) -> None:  # => co-14: builds the SAME deterministic 25-row
    # => dataset used by every pagination/filter/sort example in this topic, so worked-out ids
    # => like "done={2,5,8,...}" stay correct however many of these example directories exist
    statuses = ["todo", "in_progress", "done"]  # => rotates every 3 rows -- index is i % 3
    # => this example's filter is on THIS field, so its exact rotation is what makes counts checkable
    priorities = ["low", "normal", "high"]  # => rotates every 2 rows via (i // 2) % 3 --
    # => DECORRELATED from status ON PURPOSE, so a combined status+priority filter genuinely
    # => narrows further than either field alone, instead of the two columns moving in lockstep
    rows = [
        (f"task {i}", statuses[i % 3], priorities[(i // 2) % 3], f"2026-07-{i:02d}T00:00:00")
        for i in range(1, 26)  # => same 25-row seed -- status=done matches ids {2,5,8,11,14,17,20,23}
        # => (8 of the 25 rows -- verified by counting i % 3 == 2 for i in 1..25, since "done" is index 2)
    ]
    conn.executemany(  # => co-14: a single batched INSERT for all 25 rows, one transaction
        "INSERT INTO tasks (title, status, priority, created_at) VALUES (?, ?, ?, ?)",  # => co-14:
        # => ? placeholders -- the SAME parameterization principle Example 71 stress-tests directly
        rows,
    )
    conn.commit()  # => co-14: commits the whole seed batch at once, not row-by-row
 
 
def init_db() -> None:  # => co-15: fresh schema + seed data every time this module is imported
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)  # => start from a known, deterministic state -- this example's expected
        # => row counts and ids only hold true starting from a CLEAN file, never an accumulated one
    conn = sqlite3.connect(DB_PATH)
    conn.execute(SCHEMA)  # => co-15: creates the table every query below depends on existing
    _seed(conn)  # => populates it with the 25-row deterministic dataset described above
    conn.close()  # => co-14: connections are short-lived here -- opened, used, closed, never held
 
 
def get_connection() -> sqlite3.Connection:  # => co-14: the repository's ONLY entry point to
    # => the DB -- every query function below calls THIS instead of sqlite3.connect() directly
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # => rows behave like dicts -- readable by column name, which
    # => is what makes `dict(row)` below produce sensible {"id":..., "title":...} keys, not a tuple
    return conn  # => a fresh connection per call -- co-14: no pooling, matches this example's scale
 
 
def list_tasks(status: str | None) -> list[TaskRow]:  # => co-14, co-20: the repository's filter
    # => parameter -- None means "no filter at all," a real value means "narrow to exactly this"
    conn = get_connection()
    if status is None:  # => co-20: no filter -- the FULL 25-row table, same query as Example 65
        cursor = conn.execute("SELECT id, title, status, priority, created_at FROM tasks ORDER BY id")
    else:  # => co-14, co-20: a parameterized WHERE clause -- never string-formatted into the SQL,
        # => which is exactly what keeps a value like `done' OR '1'='1` (Example 71) inert here too
        cursor = conn.execute(
            "SELECT id, title, status, priority, created_at FROM tasks WHERE status = ? ORDER BY id",
            (status,),  # => co-14: the single ? placeholder -- SQLite treats this as DATA, never SQL
        )
    result = [dict(row) for row in cursor.fetchall()]  # type: ignore[misc]  # => sqlite3.Row -> dict
    conn.close()  # => co-14: closed immediately after whichever branch above ran
    return result  # type: ignore[return-value]  # => shape matches TaskRow at runtime
 
 
init_db()  # => co-15: runs once at import time, before the app starts serving
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the filter mechanism
 
 
@app.get("/tasks")  # => co-20: this example's focus -- a single optional filter query param --
# => co-20: `status=done` matches exactly the 8 seeded rows at ids {2,5,8,11,14,17,20,23}
def get_tasks(status: str | None = Query(default=None)) -> list[TaskRow]:
    return list_tasks(status)  # => co-24: the handler holds no SQL -- it delegates to the repository

=== GET /tasks (unfiltered) ===
row count: 25
=== GET /tasks?status=done ===
[{"id":2,"title":"task 2","status":"done","priority":"normal","created_at":"2026-07-02T00:00:00"},{"id":5,"title":"task 5","status":"done","priority":"high","created_at":"2026-07-05T00:00:00"},{"id":8,"title":"task 8","status":"done","priority":"normal","created_at":"2026-07-08T00:00:00"},{"id":11,"title":"task 11","status":"done","priority":"high","created_at":"2026-07-11T00:00:00"},{"id":14,"title":"task 14","status":"done","priority":"normal","created_at":"2026-07-14T00:00:00"},{"id":17,"title":"task 17","status":"done","priority":"high","created_at":"2026-07-17T00:00:00"},{"id":20,"title":"task 20","status":"done","priority":"normal","created_at":"2026-07-20T00:00:00"},{"id":23,"title":"task 23","status":"done","priority":"high","created_at":"2026-07-23T00:00:00"}]

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_no_filter_returns_all_twenty_five PASSED               [ 33%]
test_app.py::test_status_done_returns_only_eight_matching_rows PASSED    [ 66%]
test_app.py::test_status_with_no_matches_returns_empty 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.18s =========================

"""Example 70: Filter Multiple -- ?status=done&priority=high combine with AND semantics."""
# => co-20: this is exactly WHY Example 65's seed decorrelates status from priority -- the two
# => filters below only prove something new if their intersection is smaller than either alone
 
import os  # => builds DB_PATH below in an OS-independent way
import sqlite3  # => co-14: the stdlib DB driver -- no ORM, no extra dependency needed
from typing import TypedDict  # => co-09: a typed dict shape for what the repository returns
 
from fastapi import FastAPI, Query  # => co-12: query params are typed and validated here
# => (fully self-contained: nothing here is imported from any other example directory)
 
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")  # => co-14: one on-disk SQLite
# => file PER EXAMPLE DIRECTORY -- never shared with any other example, keeping each self-contained
 
# => co-15: the schema below backs every list query in this example. Column-by-column:
# =>   id          -- an auto-incrementing primary key, never chosen by the caller
# =>   title       -- the task's display text, no constraint on it beyond NOT NULL
# =>   status      -- co-20: a FILTERABLE field, exercised starting at Example 69
# =>   priority    -- co-20: a SECOND filterable field, exercised starting at Example 70
# =>   created_at  -- co-20: a SORTABLE field, exercised starting at Example 72
SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    status TEXT NOT NULL,
    priority TEXT NOT NULL,
    created_at TEXT NOT NULL
);
"""
 
 
class TaskRow(TypedDict):  # => co-14, co-24: the repository's typed return shape --
    # => every field below matches a SCHEMA column one-to-one, so `dict(row)` (further down)
    # => produces a value that satisfies this shape without any manual field-by-field mapping
    id: int  # => matches the schema's INTEGER PRIMARY KEY column
    title: str  # => matches the schema's title column
    status: str  # => matches the schema's status column
    priority: str  # => matches the schema's priority column
    created_at: str  # => matches the schema's created_at column
 
 
def _seed(conn: sqlite3.Connection) -> None:  # => co-14: builds the SAME deterministic 25-row
    # => dataset used by every pagination/filter/sort example in this topic, so worked-out ids
    # => like "done={2,5,8,...}" stay correct however many of these example directories exist
    statuses = ["todo", "in_progress", "done"]  # => rotates every 3 rows -- index is i % 3
    priorities = ["low", "normal", "high"]  # => rotates every 2 rows via (i // 2) % 3 --
    # => DECORRELATED from status ON PURPOSE, so a combined status+priority filter genuinely
    # => narrows further than either field alone, instead of the two columns moving in lockstep
    rows = [
        (f"task {i}", statuses[i % 3], priorities[(i // 2) % 3], f"2026-07-{i:02d}T00:00:00")
        for i in range(1, 26)  # => done={2,5,8,11,14,17,20,23}, high={4,5,10,11,16,17,22,23}
    ]
    conn.executemany(  # => co-14: a single batched INSERT for all 25 rows, one transaction
        "INSERT INTO tasks (title, status, priority, created_at) VALUES (?, ?, ?, ?)",  # => co-14:
        # => ? placeholders -- the SAME parameterization principle Example 71 stress-tests directly
        rows,
    )
    conn.commit()  # => co-14: commits the whole seed batch at once, not row-by-row
 
 
def init_db() -> None:  # => co-15: fresh schema + seed data every time this module is imported
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)  # => start from a known, deterministic state -- this example's expected
        # => row counts and ids only hold true starting from a CLEAN file, never an accumulated one
    conn = sqlite3.connect(DB_PATH)
    conn.execute(SCHEMA)  # => co-15: creates the table every query below depends on existing
    _seed(conn)  # => populates it with the 25-row deterministic dataset described above
    conn.close()  # => co-14: connections are short-lived here -- opened, used, closed, never held
 
 
def get_connection() -> sqlite3.Connection:  # => co-14: the repository's ONLY entry point to
    # => the DB -- every query function below calls THIS instead of sqlite3.connect() directly
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # => rows behave like dicts -- readable by column name, which
    # => is what makes `dict(row)` below produce sensible {"id":..., "title":...} keys, not a tuple
    return conn  # => a fresh connection per call -- co-14: no pooling, matches this example's scale
 
 
def list_tasks(status: str | None, priority: str | None) -> list[TaskRow]:  # => co-20: TWO
    # => independent filters -- each optional on its own, but AND-combined when both are present
    conn = get_connection()
    clauses: list[str] = []  # => co-14, co-20: builds the WHERE clause incrementally, still
    # => fully parameterized -- ONLY the placeholder count/order is dynamic, never the SQL text itself
    params: list[str] = []  # => co-14: kept in lockstep with `clauses` -- one entry per ? placeholder
    if status is not None:  # => co-20: opt-in -- omitting ?status= entirely skips this clause completely
        clauses.append("status = ?")  # => placeholder ONLY -- never f"status = '{status}'"
        params.append(status)  # => co-14: the actual value, bound positionally to the ? above
    if priority is not None:  # => co-20: independently opt-in -- can combine with status or stand alone
        clauses.append("priority = ?")  # => a SECOND placeholder, combined with AND below
        params.append(priority)  # => co-14: appended AFTER status's param, matching clause order
    where = f" WHERE {' AND '.join(clauses)}" if clauses else ""  # => co-20: AND joins every active
    # => filter -- zero filters means an empty string here, reproducing Example 65's unfiltered query
    cursor = conn.execute(  # => co-14: only the CLAUSE STRUCTURE is built dynamically; every value
        # => still flows through params as a placeholder, so injection is impossible regardless of input
        f"SELECT id, title, status, priority, created_at FROM tasks{where} ORDER BY id",  # => co-20
        params,  # => co-14: 0, 1, or 2 bound values, matching however many clauses were built above
    )
    result = [dict(row) for row in cursor.fetchall()]  # type: ignore[misc]  # => sqlite3.Row -> dict
    conn.close()  # => co-14: closed immediately after this one query
    return result  # type: ignore[return-value]  # => shape matches TaskRow at runtime
 
 
init_db()  # => co-15: runs once at import time, before the app starts serving
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the combined-filter mechanism
 
 
@app.get("/tasks")  # => co-20: this example's focus -- TWO filters combined with AND semantics
def get_tasks(
    status: str | None = Query(default=None),  # => co-12: independently optional, same as Example 69
    priority: str | None = Query(default=None),  # => co-12: the SECOND independently optional filter
) -> list[TaskRow]:
    return list_tasks(status, priority)  # => co-24: the handler holds no SQL, only orchestration

=== GET /tasks?status=done (single filter) ===
row count: 8
ids: [2, 5, 8, 11, 14, 17, 20, 23]
=== GET /tasks?status=done&priority=high (combined AND filter) ===
[{"id":5,"title":"task 5","status":"done","priority":"high","created_at":"2026-07-05T00:00:00"},{"id":11,"title":"task 11","status":"done","priority":"high","created_at":"2026-07-11T00:00:00"},{"id":17,"title":"task 17","status":"done","priority":"high","created_at":"2026-07-17T00:00:00"},{"id":23,"title":"task 23","status":"done","priority":"high","created_at":"2026-07-23T00:00:00"}]

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 2 items
 
test_app.py::test_status_alone_returns_eight PASSED                      [ 50%]
test_app.py::test_status_and_priority_together_narrow_further 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 71: Filter Parameterized SQL -- an injection attempt is neutralized."""
# => co-14, co-20: every other pagination/filter example in this topic ALREADY parameterizes
# => its SQL -- this example is the one that proves WHY, by showing the vulnerable alternative
 
import os
import sqlite3  # => co-14: the stdlib DB driver -- no ORM, no extra dependency needed
from typing import TypedDict  # => co-09: a typed dict shape for what the repository returns
 
from fastapi import FastAPI, Query  # => co-12: query params are typed and validated here
# => (fully self-contained: nothing here is imported from any other example directory)
 
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")  # => co-14: one on-disk SQLite
# => file PER EXAMPLE DIRECTORY -- never shared with any other example, keeping each self-contained
 
# => co-15: the schema below backs every list query in this example. Column-by-column:
# =>   id          -- an auto-incrementing primary key, never chosen by the caller
# =>   title       -- the task's display text, no constraint on it beyond NOT NULL
# =>   status      -- co-20: a FILTERABLE field, exercised starting at Example 69
# =>   priority    -- co-20: a SECOND filterable field, exercised starting at Example 70
# =>   created_at  -- co-20: a SORTABLE field, exercised starting at Example 72
SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    status TEXT NOT NULL,
    priority TEXT NOT NULL,
    created_at TEXT NOT NULL
);
"""
 
 
class TaskRow(TypedDict):  # => co-14, co-24: the repository's typed return shape --
    # => every field below matches a SCHEMA column one-to-one, so `dict(row)` (further down)
    # => produces a value that satisfies this shape without any manual field-by-field mapping
    id: int  # => matches the schema's INTEGER PRIMARY KEY column
    title: str  # => matches the schema's title column
    status: str  # => matches the schema's status column
    priority: str  # => matches the schema's priority column
    created_at: str  # => matches the schema's created_at column
 
 
def _seed(conn: sqlite3.Connection) -> None:  # => co-14: builds the SAME deterministic 25-row
    # => dataset used by every pagination/filter/sort example in this topic, so worked-out ids
    # => like "done={2,5,8,...}" stay correct however many of these example directories exist
    statuses = ["todo", "in_progress", "done"]  # => rotates every 3 rows -- index is i % 3
    priorities = ["low", "normal", "high"]  # => rotates every 2 rows via (i // 2) % 3 --
    # => DECORRELATED from status ON PURPOSE, so a combined status+priority filter genuinely
    # => narrows further than either field alone, instead of the two columns moving in lockstep
    rows = [(f"task {i}", statuses[i % 3], priorities[(i // 2) % 3], f"2026-07-{i:02d}T00:00:00") for i in range(1, 26)]
    conn.executemany(  # => co-14: a single batched INSERT for all 25 rows, one transaction
        "INSERT INTO tasks (title, status, priority, created_at) VALUES (?, ?, ?, ?)",  # => co-14:
        # => ? placeholders -- the SAME parameterization principle Example 71 stress-tests directly
        rows,
    )
    conn.commit()  # => co-14: commits the whole seed batch at once, not row-by-row
 
 
def init_db() -> None:  # => co-15: fresh schema + seed data every time this module is imported
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)  # => start from a known, deterministic state -- this example's expected
        # => row counts and ids only hold true starting from a CLEAN file, never an accumulated one
    conn = sqlite3.connect(DB_PATH)
    conn.execute(SCHEMA)  # => co-15: creates the table every query below depends on existing
    _seed(conn)  # => populates it with the 25-row deterministic dataset described above
    conn.close()  # => co-14: connections are short-lived here -- opened, used, closed, never held
 
 
def get_connection() -> sqlite3.Connection:  # => co-14: the repository's ONLY entry point to
    # => the DB -- every query function below calls THIS instead of sqlite3.connect() directly
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # => rows behave like dicts -- readable by column name, which
    # => is what makes `dict(row)` below produce sensible {"id":..., "title":...} keys, not a tuple
    return conn  # => a fresh connection per call -- co-14: no pooling, matches this example's scale
 
 
def list_tasks_safe(status: str) -> list[TaskRow]:  # => co-14, co-20: the SAFE, parameterized version --
    # => this is the ONLY function this example's route actually calls, further down
    conn = get_connection()
    cursor = conn.execute(  # => co-14: the value is bound as a PARAMETER, never concatenated into the SQL
        "SELECT id, title, status, priority, created_at FROM tasks WHERE status = ? ORDER BY id",
        (status,),  # => sqlite3 sends this as DATA, not as part of the SQL grammar at all -- an
        # => injection payload like `done' OR '1'='1` is just a literal STRING being compared here
    )
    result = [dict(row) for row in cursor.fetchall()]  # type: ignore[misc]  # => sqlite3.Row -> dict
    conn.close()  # => co-14: closed immediately after this one query
    return result  # type: ignore[return-value]  # => empty list when no row's status equals the payload
 
 
def list_tasks_unsafe_for_demo_only(  # => co-14: the VULNERABLE version -- exists ONLY as a
    status: str,  # => teaching contrast; never called by any route, only by this example's own tests
) -> list[TaskRow]:
    conn = get_connection()
    query = (  # => co-14: string-BUILDING the SQL text itself, not just its parameters
        f"SELECT id, title, status, priority, created_at FROM tasks WHERE status = '{status}' ORDER BY id"
    )
    # => DELIBERATELY f-string-interpolated -- shown ONLY to prove what parameterization prevents,
    # => never something a real handler should do; not wired to any route in this app
    cursor = conn.execute(query)  # => co-14: SQLite parses whatever ends up embedded in `query` as SQL
    result = [dict(row) for row in cursor.fetchall()]  # type: ignore[misc]  # => sqlite3.Row -> dict
    conn.close()  # => co-14: closed immediately after this one query
    return result  # type: ignore[return-value]  # => ALL 25 rows when the payload closes the quote early
 
 
init_db()  # => co-15: runs once at import time, before the app starts serving
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the safe/unsafe query contrast
 
 
@app.get("/tasks")  # => co-20: the ONLY route this app exposes uses the SAFE repository function --
# => `list_tasks_unsafe_for_demo_only` above is reachable only from Python code (like the tests
# => and curl-equivalent script below), never from any HTTP route a real caller could hit
def get_tasks(status: str = Query(...)) -> list[TaskRow]:  # => co-12: required, no default -- omitting
    # => `?status=` entirely is a 422, distinct from THIS example's own scenario of a crafted value
    return list_tasks_safe(status)  # => co-24: the handler always goes through the SAFE path

=== GET /tasks?status=done' OR '1'='1 (SQL injection attempt, parameterized route) ===
[]
=== GET /tasks?status=done (legitimate filter, for contrast) ===
row count: 8

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 4 items
 
test_app.py::test_route_neutralizes_the_injection_attempt PASSED         [ 25%]
test_app.py::test_safe_repository_function_matches_route_behavior PASSED [ 50%]
test_app.py::test_unsafe_function_demonstrates_the_vulnerability_it_would_have_had PASSED [ 75%]
test_app.py::test_safe_status_lookup_still_works_normally 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
========================= 4 passed, 1 warning in 0.16s =========================

"""Example 72: Sort Param -- ?sort=created_at (or -created_at) orders the list."""
 
import os
import sqlite3  # => co-14: the stdlib DB driver -- no ORM, no extra dependency needed
from typing import Literal, TypedDict  # => co-12: Literal constrains the sort param to two values
 
from fastapi import FastAPI, Query  # => co-12: query params are typed and validated here
# => (fully self-contained: nothing here is imported from any other example directory)
 
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")  # => co-14: one on-disk SQLite
# => file PER EXAMPLE DIRECTORY -- never shared with any other example, keeping each self-contained
 
# => co-15: the schema below backs every list query in this example. Column-by-column:
# =>   id          -- an auto-incrementing primary key, never chosen by the caller
# =>   title       -- the task's display text, no constraint on it beyond NOT NULL
# =>   status      -- co-20: a FILTERABLE field, exercised starting at Example 69
# =>   priority    -- co-20: a SECOND filterable field, exercised starting at Example 70
# =>   created_at  -- co-20: a SORTABLE field, exercised starting at Example 72
SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    status TEXT NOT NULL,
    priority TEXT NOT NULL,
    created_at TEXT NOT NULL
);
"""
 
 
class TaskRow(TypedDict):  # => co-14, co-24: the repository's typed return shape --
    # => every field below matches a SCHEMA column one-to-one, so `dict(row)` (further down)
    # => produces a value that satisfies this shape without any manual field-by-field mapping
    id: int  # => matches the schema's INTEGER PRIMARY KEY column
    title: str  # => matches the schema's title column
    status: str  # => matches the schema's status column
    priority: str  # => matches the schema's priority column
    created_at: str  # => matches the schema's created_at column
 
 
def _seed(conn: sqlite3.Connection) -> None:  # => co-14: builds the SAME deterministic 25-row
    # => dataset used by every pagination/filter/sort example in this topic, so worked-out ids
    # => like "done={2,5,8,...}" stay correct however many of these example directories exist
    statuses = ["todo", "in_progress", "done"]  # => rotates every 3 rows -- index is i % 3
    priorities = ["low", "normal", "high"]  # => rotates every 2 rows via (i // 2) % 3 --
    # => DECORRELATED from status ON PURPOSE, so a combined status+priority filter genuinely
    # => narrows further than either field alone, instead of the two columns moving in lockstep
    rows = [
        (f"task {i}", statuses[i % 3], priorities[(i // 2) % 3], f"2026-07-{i:02d}T00:00:00")
        for i in range(1, 26)  # => created_at already increases WITH id -- ascending sort looks
        # => like a no-op, so THIS example's curl deliberately checks descending order too, below
    ]
    conn.executemany(  # => co-14: a single batched INSERT for all 25 rows, one transaction
        "INSERT INTO tasks (title, status, priority, created_at) VALUES (?, ?, ?, ?)",  # => co-14:
        # => ? placeholders -- the SAME parameterization principle Example 71 stress-tests directly
        rows,
    )
    conn.commit()  # => co-14: commits the whole seed batch at once, not row-by-row
 
 
def init_db() -> None:  # => co-15: fresh schema + seed data every time this module is imported
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)  # => start from a known, deterministic state -- this example's expected
        # => row counts and ids only hold true starting from a CLEAN file, never an accumulated one
    conn = sqlite3.connect(DB_PATH)
    conn.execute(SCHEMA)  # => co-15: creates the table every query below depends on existing
    _seed(conn)  # => populates it with the 25-row deterministic dataset described above
    conn.close()  # => co-14: connections are short-lived here -- opened, used, closed, never held
 
 
def get_connection() -> sqlite3.Connection:  # => co-14: the repository's ONLY entry point to
    # => the DB -- every query function below calls THIS instead of sqlite3.connect() directly
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # => rows behave like dicts -- readable by column name, which
    # => is what makes `dict(row)` below produce sensible {"id":..., "title":...} keys, not a tuple
    return conn  # => a fresh connection per call -- co-14: no pooling, matches this example's scale
 
 
SortValue = Literal["created_at", "-created_at"]  # => co-20: a closed set -- FastAPI 422s anything
# => else, unlike `status`/`priority` in the filter examples, which accept any string at all
 
_COLUMN_BY_SORT: dict[SortValue, str] = {  # => co-14: maps the PUBLIC sort value to a real column +
    # => direction -- this dict is the ONLY thing standing between user input and an ORDER BY clause
    "created_at": "created_at ASC",  # => oldest-first, matching the seed's natural insertion order
    "-created_at": "created_at DESC",  # => a leading "-" is a common REST convention for "descending"
}
 
 
def list_tasks(sort: SortValue) -> list[TaskRow]:  # => co-14, co-20: sort is a Literal, so `sort`
    # => can ONLY ever be one of the two dict keys above -- pyright itself enforces total coverage
    conn = get_connection()
    order_clause = _COLUMN_BY_SORT[sort]  # => co-14: looked up from a FIXED mapping -- never
    # => interpolates raw user input directly into ORDER BY, which parameterization cannot protect
    # => against (SQLite has no `?` placeholder syntax for column names or ASC/DESC keywords at all)
    cursor = conn.execute(f"SELECT id, title, status, priority, created_at FROM tasks ORDER BY {order_clause}")
    result = [dict(row) for row in cursor.fetchall()]  # type: ignore[misc]  # => sqlite3.Row -> dict
    conn.close()  # => co-14: closed immediately after this one query
    return result  # type: ignore[return-value]  # => shape matches TaskRow at runtime
 
 
init_db()  # => co-15: runs once at import time, before the app starts serving
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the sort mechanism
 
 
@app.get("/tasks")  # => co-20: this example's focus -- reordering the SAME underlying rows,
# => never changing WHICH rows come back, only the SEQUENCE they arrive in
def get_tasks(sort: SortValue = Query(default="created_at")) -> list[TaskRow]:
    return list_tasks(sort)  # => co-24: the handler holds no SQL -- it delegates to the repository

=== GET /tasks?sort=created_at (ascending) ===
first 3 ids: [1, 2, 3]
last 3 ids: [23, 24, 25]
=== GET /tasks?sort=-created_at (descending) ===
first 3 ids: [25, 24, 23]
last 3 ids: [3, 2, 1]

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_ascending_sort_starts_at_id_1 PASSED                   [ 33%]
test_app.py::test_descending_sort_reverses_the_order PASSED              [ 66%]
test_app.py::test_invalid_sort_value_is_422 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.19s =========================

"""Example 73: Combined List Query -- pagination + filter + sort together."""
# => co-19, co-20: Examples 65-72 built pagination, filtering, and sorting SEPARATELY -- this
# => example is the capstone-style proof that all three compose correctly in a single endpoint,
# => without one feature silently breaking another (e.g. total must reflect the FILTER, not ignore it)
 
import os  # => co-14: used for the DB_PATH lookup and the "start fresh" file check below
import sqlite3  # => co-14: the stdlib DB driver -- no ORM, no extra dependency needed
from typing import Literal, TypedDict  # => co-12: Literal constrains the sort param to two values
 
from fastapi import FastAPI, Query  # => co-12: query params are typed and validated here
# => (fully self-contained: nothing here is imported from any other example directory)
 
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")  # => co-14: one on-disk SQLite
# => file PER EXAMPLE DIRECTORY -- never shared with any other example, keeping each self-contained
 
# => co-15: the schema below backs every list query in this example. Column-by-column:
# =>   id          -- an auto-incrementing primary key, never chosen by the caller
# =>   title       -- the task's display text, no constraint on it beyond NOT NULL
# =>   status      -- co-20: a FILTERABLE field, exercised starting at Example 69
# =>   priority    -- co-20: a SECOND filterable field, exercised starting at Example 70
# =>   created_at  -- co-20: a SORTABLE field, exercised starting at Example 72
SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    status TEXT NOT NULL,
    priority TEXT NOT NULL,
    created_at TEXT NOT NULL
);
"""
 
 
class TaskRow(TypedDict):  # => co-14, co-24: the repository's typed return shape --
    # => every field below matches a SCHEMA column one-to-one, so `dict(row)` (further down)
    # => produces a value that satisfies this shape without any manual field-by-field mapping
    id: int  # => matches the schema's INTEGER PRIMARY KEY column
    title: str  # => matches the schema's title column
    status: str  # => matches the schema's status column
    priority: str  # => matches the schema's priority column
    created_at: str  # => matches the schema's created_at column
 
 
class Page(TypedDict):  # => co-09, co-19: the SAME envelope shape as Example 67, reused here
    items: list[TaskRow]  # => this page's rows, after filtering AND sorting AND slicing
    total: int  # => co-19, co-20: the FILTERED count -- not the whole 25-row table
    next: int | None  # => co-19: computed against the filtered total, not the unfiltered one
 
 
def _seed(conn: sqlite3.Connection) -> None:  # => co-14: builds the SAME deterministic 25-row
    # => dataset used by every pagination/filter/sort example in this topic, so worked-out ids
    # => like "done={2,5,8,...}" stay correct however many of these example directories exist
    statuses = ["todo", "in_progress", "done"]  # => rotates every 3 rows -- index is i % 3
    priorities = ["low", "normal", "high"]  # => rotates every 2 rows via (i // 2) % 3 --
    # => DECORRELATED from status ON PURPOSE, so a combined status+priority filter genuinely
    # => narrows further than either field alone, instead of the two columns moving in lockstep
    rows = [
        (f"task {i}", statuses[i % 3], priorities[(i // 2) % 3], f"2026-07-{i:02d}T00:00:00")
        for i in range(1, 26)  # => the SAME 25-row seed as every other pagination/filter/sort example
    ]
    conn.executemany(  # => co-14: a single batched INSERT for all 25 rows, one transaction
        "INSERT INTO tasks (title, status, priority, created_at) VALUES (?, ?, ?, ?)",  # => co-14:
        # => ? placeholders -- the SAME parameterization principle Example 71 stress-tests directly
        rows,
    )
    conn.commit()  # => co-14: commits the whole seed batch at once, not row-by-row
 
 
def init_db() -> None:  # => co-15: fresh schema + seed data every time this module is imported
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)  # => start from a known, deterministic state -- this example's expected
        # => row counts and ids only hold true starting from a CLEAN file, never an accumulated one
    conn = sqlite3.connect(DB_PATH)
    conn.execute(SCHEMA)  # => co-15: creates the table every query below depends on existing
    _seed(conn)  # => populates it with the 25-row deterministic dataset described above
    conn.close()  # => co-14: connections are short-lived here -- opened, used, closed, never held
 
 
def get_connection() -> sqlite3.Connection:  # => co-14: the repository's ONLY entry point to
    # => the DB -- every query function below calls THIS instead of sqlite3.connect() directly
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # => rows behave like dicts -- readable by column name, which
    # => is what makes `dict(row)` below produce sensible {"id":..., "title":...} keys, not a tuple
    return conn  # => a fresh connection per call -- co-14: no pooling, matches this example's scale
 
 
SortValue = Literal["created_at", "-created_at"]  # => co-20: same closed set as Example 72
_COLUMN_BY_SORT: dict[SortValue, str] = {  # => co-14: the SAME lookup-not-interpolate pattern
    "created_at": "created_at ASC",
    "-created_at": "created_at DESC",
}
 
 
def list_page(  # => co-19, co-20: ALL THREE features compose in one repository function --
    # => the order below matters: filter narrows the ROWS, sort orders them, THEN limit/offset slices
    status: str | None,  # => co-20: optional -- None means every status qualifies
    limit: int,  # => co-19: how many rows this page holds, at most
    offset: int,  # => co-19: how many filtered-and-sorted rows to skip before this page starts
    sort: SortValue,  # => co-20: which column/direction orders the result, before slicing happens
) -> Page:
    conn = get_connection()  # => co-14: one connection, reused for both queries this function issues
    clauses: list[str] = []  # => co-14, co-20: the filter half -- still fully parameterized,
    # => identical technique to Example 69, just reused inside a function that also sorts and paginates
    params: list[str] = []  # => co-14: kept in lockstep with `clauses`, exactly like Example 70
    if status is not None:  # => co-20: opt-in, exactly like the standalone filter example
        clauses.append("status = ?")  # => the one filterable column this composed example supports
        params.append(status)  # => co-14: the bound value, positionally matched to the ? above
    where = f" WHERE {' AND '.join(clauses)}" if clauses else ""  # => empty string when unfiltered
 
    total = int(conn.execute(f"SELECT COUNT(*) FROM tasks{where}", params).fetchone()[0])  # => co-19: total reflects the FILTERED count, not the whole table -- computed with the
    # => SAME where clause and params as the main query below, so the two numbers stay consistent
    order_clause = _COLUMN_BY_SORT[sort]  # => co-20: the sort half -- looked up, never interpolated raw
    cursor = conn.execute(  # => co-14, co-19, co-20: filter, sort, AND paginate in one statement
        f"SELECT id, title, status, priority, created_at FROM tasks{where} ORDER BY {order_clause} LIMIT ? OFFSET ?",  # => co-19: the pagination half, composed last --
        # => SQL applies WHERE, then ORDER BY, then LIMIT/OFFSET, in that fixed evaluation order
        [*params, limit, offset],  # => co-14: filter params FIRST, then limit/offset -- position
        # => must match the ? placeholders' left-to-right order in the SQL string exactly above
    )
    items = [dict(row) for row in cursor.fetchall()]  # type: ignore[misc]  # => sqlite3.Row -> dict
    conn.close()  # => co-14: closed immediately after both queries above finish
    next_offset = offset + limit  # => co-19: identical formula to Example 67, applied to the FILTERED total
    has_more = next_offset < total  # => whether another filtered-and-sorted page exists past this one
    return {  # => co-19, co-20: the SAME Page envelope shape as Example 67, now built from a
        # => query that filtered, sorted, AND paginated in a single round trip to SQLite
        "items": items,  # type: ignore[typeddict-item]  # => this page's slice, after all three steps
        "total": total,  # => co-19, co-20: the FILTERED total -- proves filter+pagination interact correctly
        "next": next_offset if has_more else None,  # => None once the filtered result set is exhausted
    }
 
 
init_db()  # => co-15: runs once at import time, before the app starts serving
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the composed list mechanism
 
 
@app.get("/tasks")  # => co-19, co-20: this example's focus -- all three composed in one call --
# => every query param below is INDEPENDENTLY optional, exactly matching its own standalone example
def get_tasks(
    status: str | None = Query(default=None),  # => co-20: same as Example 69
    limit: int = Query(default=10, ge=1, le=50),  # => co-19: same bounds as Example 68
    offset: int = Query(default=0, ge=0),  # => co-19: same as Example 65
    sort: SortValue = Query(default="created_at"),  # => co-20: same as Example 72
) -> Page:
    # => co-24: four independently-optional parameters, one delegated call -- zero SQL in this handler
    return list_page(status, limit, offset, sort)  # => co-24: the handler holds no SQL at all

=== GET /tasks?status=done&limit=3&offset=0&sort=-created_at (all three composed) ===
{"items":[{"id":23,"title":"task 23","status":"done","priority":"high","created_at":"2026-07-23T00:00:00"},{"id":20,"title":"task 20","status":"done","priority":"normal","created_at":"2026-07-20T00:00:00"},{"id":17,"title":"task 17","status":"done","priority":"high","created_at":"2026-07-17T00:00:00"}],"total":8,"next":3}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_filter_alone_reports_the_filtered_total PASSED         [ 33%]
test_app.py::test_filter_plus_pagination_plus_sort_all_compose PASSED    [ 66%]
test_app.py::test_second_page_of_the_same_filtered_sorted_query 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 74: Idempotent PUT, Verified -- the same PUT applied twice leaves the same state."""
# => co-02: RFC 9110 CLASSIFIES PUT as idempotent, but a classification is only a promise --
# => this example actually PROVES it, by calling PUT twice and checking the row count stays 1
 
import os  # => co-14: used for the DB_PATH lookup and the "start fresh" file check below
import sqlite3  # => co-14: the stdlib DB driver -- no ORM, no extra dependency needed
from typing import TypedDict  # => co-09: a typed dict shape for what the repository returns
 
from fastapi import FastAPI, HTTPException  # => co-03: HTTPException raises the 404 branch below
from pydantic import BaseModel  # => co-10: validates the PUT request body's shape
# => (fully self-contained: nothing here is imported from any other example directory)
 
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db")  # => co-14: one on-disk SQLite file,
# => PER EXAMPLE DIRECTORY -- never shared with any other example, keeping this one self-contained
 
SCHEMA = """
CREATE TABLE IF NOT EXISTS tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    status TEXT NOT NULL
);
"""  # => co-15: a DELIBERATELY smaller schema than the pagination examples -- this one is about
# => idempotency, not filtering/sorting, so priority/created_at columns would only be noise here
 
 
class TaskRow(TypedDict):  # => co-14, co-24: the repository's typed return shape
    id: int  # => matches the schema's INTEGER PRIMARY KEY column
    title: str  # => matches the schema's title column
    status: str  # => matches the schema's status column
 
 
class TaskUpdate(BaseModel):  # => co-02, co-10: PUT REPLACES the full resource with this exact shape --
    # => unlike PATCH, there is no "only send the fields you're changing" option with PUT
    title: str  # => required -- omitting it entirely is a 422, not a partial update
    status: str  # => required -- same rule as title above
 
 
def init_db() -> None:  # => co-15: fresh schema + one seed row every time this module is imported
    if os.path.exists(DB_PATH):
        os.remove(DB_PATH)  # => start from a known, deterministic state every run
    conn = sqlite3.connect(DB_PATH)  # => co-14: a plain connection -- no row_factory needed for INSERT
    conn.execute(SCHEMA)  # => co-15: creates the table every query below depends on existing
    conn.execute(  # => co-14: a single-row INSERT, parameterized like every write in this topic
        "INSERT INTO tasks (title, status) VALUES (?, ?)", ("write the report", "todo")
    )  # => a single seed row -- id 1, the ONLY row this example's PUT calls will ever target
    conn.commit()  # => co-14: commits the seed insert
    conn.close()  # => co-14: connections are short-lived here -- opened, used, closed, never held
 
 
def get_connection() -> sqlite3.Connection:  # => co-14: the repository's ONLY entry point to the DB
    conn = sqlite3.connect(DB_PATH)  # => co-14: every read/write function below calls THIS helper
    conn.row_factory = sqlite3.Row  # => rows behave like dicts -- readable by column name
    return conn  # => a fresh connection per call -- co-14: no pooling, matches this example's scale
 
 
def replace_task(task_id: int, update: TaskUpdate) -> TaskRow:  # => co-02, co-14: PUT's repository
    # => function -- calling this TWICE with the SAME arguments must leave the SAME final state
    conn = get_connection()  # => co-14: one connection, used for the UPDATE and the confirming SELECT
    cursor = conn.execute(  # => co-14: cursor.rowcount tells us WHETHER a row actually matched below
        "UPDATE tasks SET title = ?, status = ? WHERE id = ?",  # => co-14: parameterized, idempotent
        # => by design -- an UPDATE that sets a column to the value it ALREADY holds changes nothing
        (update.title, update.status, task_id),  # => co-14: title, status, THEN the WHERE id last
    )
    # => co-02: nothing above distinguishes "first call" from "second identical call" -- the SQL
    # => itself has no memory of prior invocations, which is precisely what makes it idempotent
    if cursor.rowcount == 0:  # => co-02: RFC 9110 -- PUT may also CREATE; this example only replaces,
        # => so a missing id is treated as a 404 rather than silently inserting a NEW row at that id
        conn.close()  # => co-14: close before raising -- never leak a connection on the error path
        raise HTTPException(status_code=404, detail={"error": {"code": "not_found", "message": "no such task"}})
    conn.commit()  # => co-14: commits the UPDATE -- both the first AND second identical PUT commit here
    row = conn.execute(  # => co-14: re-reads the row to return its CURRENT state to the caller
        "SELECT id, title, status FROM tasks WHERE id = ?", (task_id,)
    ).fetchone()
    # => co-02: the SECOND PUT's response body is BYTE-IDENTICAL to the first one's -- proof enough
    conn.close()  # => co-14: closed after the confirming SELECT above returns the fresh row
    return dict(row)  # type: ignore[return-value]  # => sqlite3.Row -> TaskRow-shaped dict
 
 
def count_tasks() -> int:  # => proves PUT never INSERTS a duplicate row -- used only to verify
    # => idempotency, never called by any route a real client would hit
    conn = get_connection()  # => co-14: a fresh, short-lived connection just for this one scalar query
    total = conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0]  # => co-14: a single scalar query
    conn.close()  # => co-14: closed immediately after reading the single count value
    return int(total)  # => stays 1 no matter how many times /tasks/1 is PUT to below
    # => a NON-idempotent bug here would look like: POST-style INSERT instead of UPDATE, growing this
 
 
init_db()  # => co-15: runs once at import time, before the app starts serving
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the idempotency proof
# => (co-24: routes, repository, and models all live in this ONE file, deliberately not split up)
 
 
@app.put("/tasks/{task_id}")  # => co-02: PUT -- RFC 9110 classifies this method as IDEMPOTENT --
# => this route's curl below calls it TWICE with an identical body to make that promise concrete
def put_task(  # => co-02: task_id comes from the URL PATH, the rest of the state from the BODY
    task_id: int, update: TaskUpdate
) -> TaskRow:
    return replace_task(task_id, update)  # => co-24: the handler holds no SQL -- co-24 layering
 
 
@app.get("/tasks/count")  # => a helper endpoint used ONLY to prove no duplicate row was ever created
def get_count() -> dict[str, int]:  # => co-02: called before AND after the two PUTs to compare
    return {"total": count_tasks()}  # => co-02: expected to read 1, both before AND after two PUTs

=== GET /tasks/count (before) ===
{"total":1}
=== PUT /tasks/1 (first call) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:56:30 GMT
server: uvicorn
content-length: 58
content-type: application/json
 
{"id":1,"title":"write the report","status":"in_progress"}
=== PUT /tasks/1 (second, identical call) ===
HTTP/1.1 200 OK
date: Tue, 14 Jul 2026 10:56:30 GMT
server: uvicorn
content-length: 58
content-type: application/json
 
{"id":1,"title":"write the report","status":"in_progress"}
=== GET /tasks/count (after both PUTs) ===
{"total":1}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_first_put_replaces_the_resource PASSED                 [ 33%]
test_app.py::test_second_identical_put_produces_the_same_state_and_no_duplicate PASSED [ 66%]
test_app.py::test_put_missing_id_is_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.19s =========================

"""Example 75: Method Not Allowed -- an unsupported method returns 405 with an Allow header."""
# => co-03: FastAPI/Starlette handle THIS status code entirely automatically -- there is no
# => `raise HTTPException(405)` anywhere in this file; the 405 comes purely from route registration
 
from fastapi import FastAPI  # => co-16: no other import needed -- routing alone drives this example
 
app = FastAPI()  # => a fresh app -- this example needs no database, only route registration
 
 
@app.get("/tasks")  # => co-02: ONLY GET is registered for this exact path -- no PUT/POST/DELETE at all
def list_tasks() -> list[str]:
    return ["write the report"]  # => a fixed, single-item list -- this example never mutates state
 
 
@app.post("/reports")  # => a DIFFERENT path, registered with ONLY POST -- proves Allow is per-PATH
def create_report() -> dict[str, str]:
    return {"created": "true"}  # => reachable only via POST -- GET /reports itself would 405 too
 
 
# => co-03: RFC 9110 SS15.5.6 requires a 405 response to carry an Allow header listing the method(s)
# => that path genuinely supports -- Starlette (FastAPI's ASGI toolkit) generates this automatically.
# => VERIFIED NUANCE: when a single path has multiple decorators (e.g. both @app.get("/x") and
# => @app.post("/x")), Starlette's 405 Allow header reflects only the FIRST-matched route object,
# => not the union of every decorator for that path -- which is exactly why this example keeps
# => /tasks and /reports as two separate, single-method paths instead.

=== DELETE /tasks (only GET registered) ===
HTTP/1.1 405 Method Not Allowed
date: Tue, 14 Jul 2026 10:56:33 GMT
server: uvicorn
allow: GET
content-length: 31
content-type: application/json
 
{"detail":"Method Not Allowed"}
=== GET /reports (only POST registered) ===
HTTP/1.1 405 Method Not Allowed
date: Tue, 14 Jul 2026 10:56:33 GMT
server: uvicorn
allow: POST
content-length: 31
content-type: application/json
 
{"detail":"Method Not Allowed"}

============================= test session starts ==============================
plugins: anyio-4.14.2
collecting ... collected 3 items
 
test_app.py::test_delete_on_get_only_path_is_405_with_get_allow PASSED   [ 33%]
test_app.py::test_delete_on_post_only_path_is_405_with_post_allow PASSED [ 66%]
test_app.py::test_registered_methods_still_work_normally 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 76: Health vs Readiness -- /health never touches the DB, /ready pings it."""
# => co-08: two DIFFERENT questions that sound similar -- "is the process alive" (health) versus
# => "can this instance actually do its job right now" (readiness) -- orchestrators treat them
# => differently: a failed health check usually triggers a RESTART, a failed readiness check
# => just pulls the instance out of the load-balancing rotation until it recovers on its own
 
import os  # => co-14: reads the TASKS_DB_PATH env var this example's readiness check depends on
import sqlite3  # => co-14: the stdlib DB driver -- readiness pings a REAL connection, never a mock
 
from fastapi import FastAPI, Response  # => co-08: Response lets a handler override the status code
 
# => DB_PATH is read from an env var so the SAME code can point at a real file (co-14) or a
# => deliberately unreachable path -- used ONLY to genuinely simulate "the database is down"
# => without faking any response by hand
DB_PATH = os.environ.get("TASKS_DB_PATH", os.path.join(os.path.dirname(__file__), "tasks.db"))  # => co-14: no hardcoded path here at all -- this example's curl run overrides it directly
 
app = FastAPI()  # => a fresh app -- this example needs no auth, only the health/readiness contrast
 
 
def _init_db_if_reachable() -> None:  # => best-effort setup -- readiness below is what actually
    # => PROVES reachability; this function only exists so the "healthy" scenario has a real table
    try:
        conn = sqlite3.connect(DB_PATH)  # => co-14: fails immediately if DB_PATH's directory is gone
        conn.execute("CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, title TEXT NOT NULL)")
        conn.commit()  # => co-14: commits the CREATE TABLE, if the connection above succeeded at all
        conn.close()  # => co-14: closed immediately -- this function runs once, at import time
    except sqlite3.OperationalError:  # => e.g. DB_PATH points at a directory that doesn't exist
        pass  # => intentionally swallowed here -- /ready is the endpoint that surfaces this, not startup
 
 
_init_db_if_reachable()  # => co-15: runs once at import time -- may silently no-op, by design above
 
 
@app.get("/health")  # => co-08: LIVENESS -- "is the process itself running and responsive at all?"
def health() -> dict[str, str]:
    return {"status": "ok"}  # => co-03: no DB call, no dependency on anything external -- always 200
 
 
@app.get("/ready")  # => co-08, co-14: READINESS -- "can this instance actually SERVE a real request?"
def ready(response: Response) -> dict[str, str]:
    try:
        conn = sqlite3.connect(DB_PATH, timeout=1)  # => co-14: a real connection attempt, not a mock --
        # => timeout=1 keeps this check FAST, since a slow readiness probe is nearly as bad as a broken one
        conn.execute("SELECT 1")  # => the cheapest possible real query -- proves the DB genuinely answers
        conn.close()  # => co-14: closed immediately after the probe succeeds
        return {"status": "ready"}  # => co-03: 200 -- the default status FastAPI applies here
    except sqlite3.OperationalError as exc:  # => e.g. "unable to open database file"
        response.status_code = 503  # => co-03: Service Unavailable -- this instance cannot serve
        # => traffic right now; a load balancer reading THIS status is what actually acts on it
        return {"status": "not_ready", "reason": str(exc)}  # => co-11: the real underlying DB error,
        # => surfaced for operator debugging -- never shown to an end user in a production deployment