Staff Prep 11: API Design — Pagination, Filtering & Error Handling at Scale

-- Page 1: fast (reads 20 rows)
SELECT * FROM events ORDER BY created_at DESC LIMIT 20 OFFSET 0;

-- Page 100: reads 2,020 rows, returns 20
SELECT * FROM events ORDER BY created_at DESC LIMIT 20 OFFSET 2000;

-- Page 10,000: reads 200,020 rows, returns 20 — 30+ seconds
SELECT * FROM events ORDER BY created_at DESC LIMIT 20 OFFSET 200000;

-- Second problem: concurrent inserts cause row skips
-- User fetches page 1 (rows 1-20)
-- Another user inserts a new row at position 1
-- User fetches page 2 (rows 21-40 + offset) → row 21 is now at position 22 → row 20 shown twice
-- Or rows get skipped entirely

-- Cursor pagination: always efficient (uses the index)
-- First page
SELECT id, title, created_at
FROM posts
ORDER BY created_at DESC, id DESC
LIMIT 20;
-- Last row: id=5000, created_at='2026-04-03T12:00:00Z'

-- Second page: "give me rows before the cursor"
SELECT id, title, created_at
FROM posts
WHERE (created_at, id) < ('2026-04-03T12:00:00Z', 5000)
ORDER BY created_at DESC, id DESC
LIMIT 20;

-- This uses the index (created_at DESC, id DESC) efficiently
-- No matter which page number you are on, it reads exactly LIMIT rows

-- Why include id in the cursor?
-- Multiple posts can have identical created_at timestamps
-- The id makes the cursor unique and deterministic

import base64
import json
from fastapi import FastAPI, Query
from pydantic import BaseModel
from typing import Optional

def encode_cursor(created_at: str, id: int) -> str:
    payload = json.dumps({"created_at": created_at, "id": id})
    return base64.b64encode(payload.encode()).decode()

def decode_cursor(cursor: str) -> tuple[str, int]:
    payload = json.loads(base64.b64decode(cursor.encode()).decode())
    return payload["created_at"], payload["id"]

class PaginatedResponse(BaseModel):
    items: list
    next_cursor: Optional[str]
    has_more: bool

@app.get("/posts")
async def list_posts(
    cursor: Optional[str] = Query(None),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
):
    query_limit = limit + 1  # fetch one extra to determine has_more

    if cursor:
        created_at, last_id = decode_cursor(cursor)
        rows = await db.execute(
            """SELECT id, title, created_at FROM posts
               WHERE (created_at, id) < ($1::timestamptz, $2)
               ORDER BY created_at DESC, id DESC LIMIT $3""",
            created_at, last_id, query_limit
        )
    else:
        rows = await db.execute(
            """SELECT id, title, created_at FROM posts
               ORDER BY created_at DESC, id DESC LIMIT $1""",
            query_limit
        )

    posts = rows.fetchall()
    has_more = len(posts) > limit
    items = posts[:limit]  # trim the extra row

    next_cursor = None
    if has_more and items:
        last = items[-1]
        next_cursor = encode_cursor(last["created_at"].isoformat(), last["id"])

    return PaginatedResponse(items=items, next_cursor=next_cursor, has_more=has_more)

from fastapi import Query
from enum import Enum
from typing import Optional
from datetime import date

class OrderStatus(str, Enum):
    pending = "pending"
    processing = "processing"
    shipped = "shipped"
    delivered = "delivered"

@app.get("/orders")
async def list_orders(
    status: Optional[OrderStatus] = Query(None),
    user_id: Optional[int] = Query(None),
    created_after: Optional[date] = Query(None),
    created_before: Optional[date] = Query(None),
    min_amount: Optional[float] = Query(None, ge=0),
    cursor: Optional[str] = Query(None),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
):
    # Build dynamic WHERE clause safely
    conditions = []
    params = []

    if status:
        conditions.append(f"status = ${len(params)+1}")
        params.append(status.value)

    if user_id:
        conditions.append(f"user_id = ${len(params)+1}")
        params.append(user_id)

    if created_after:
        conditions.append(f"created_at >= ${len(params)+1}")
        params.append(created_after)

    if cursor:
        created_at, last_id = decode_cursor(cursor)
        conditions.append(f"(created_at, id) < (${len(params)+1}::timestamptz, ${len(params)+2})")
        params.extend([created_at, last_id])

    where = "WHERE " + " AND ".join(conditions) if conditions else ""

    sql = f"""SELECT id, user_id, amount, status, created_at
              FROM orders {where}
              ORDER BY created_at DESC, id DESC
              LIMIT ${len(params)+1}"""
    params.append(limit + 1)

    rows = await db.execute(sql, *params)
    # ... rest of pagination logic

from fastapi import Request
from fastapi.responses import JSONResponse
from pydantic import ValidationError
import uuid

# Consistent error format
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc):
    return JSONResponse(
        status_code=422,
        content={
            "error": "validation_failed",
            "message": "Request validation failed",
            "details": [
                {"field": ".".join(str(loc) for loc in e["loc"]), "message": e["msg"]}
                for e in exc.errors()
            ],
            "request_id": str(uuid.uuid4()),
        }
    )

# Clients get structured, actionable errors:
# {
#   "error": "validation_failed",
#   "message": "Request validation failed",
#   "details": [
#     {"field": "body.amount", "message": "ensure this value is greater than 0"}
#   ],
#   "request_id": "abc-123"
# }