FastAPI: Pagination, concurrency & migrations

Four patterns show up in every serious API. Cursor pagination stays fast and consistent on huge tables. Safe filtering and sorting never puts user input into SQL. Optimistic concurrency with ETags stops two users silently overwriting each other. Idempotency keys make “retry the payment” safe. Around them sits the discipline of schema migrations that do not take the service down. Each is shown running here, except Alembic, which is described and not run because it is not installed on this machine.

1. Offset vs cursor pagination

the cost of OFFSET on a deep page vs a keyset cursor (200,000 rows, SQLite)

import sqlite3
import time

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, created_at TEXT, total REAL)")
db.execute("CREATE INDEX ix_created ON orders (created_at, id)")
db.executemany("INSERT INTO orders (created_at, total) VALUES (?, ?)",
               ((f"2026-{1 + i // 20000:02d}-{1 + (i // 700) % 28:02d}T{i % 24:02d}:00", i % 997) for i in range(200_000)))

def timed(sql, params=()):
    start = time.perf_counter()
    rows = db.execute(sql, params).fetchall()
    return rows, (time.perf_counter() - start) * 1000

page_size = 20
for page in (1, 100, 9_000):
    _, ms = timed("SELECT id, created_at FROM orders ORDER BY created_at, id LIMIT ? OFFSET ?",
                  (page_size, (page - 1) * page_size))
    print(f"OFFSET page {page:>5}: {ms:7.2f} ms")

# keyset: remember the last row of the previous page and continue AFTER it
last = db.execute("SELECT created_at, id FROM orders ORDER BY created_at, id LIMIT 1 OFFSET 179999").fetchone()
rows, ms = timed("SELECT id, created_at FROM orders WHERE (created_at, id) > (?, ?) ORDER BY created_at, id LIMIT ?",
                 (*last, page_size))
print(f"cursor, same depth: {ms:7.2f} ms  (first row id {rows[0][0]})")

Output

OFFSET page     1:    0.16 ms
OFFSET page   100:    0.06 ms
OFFSET page  9000:    2.28 ms
cursor, same depth:    0.24 ms  (first row id 196009)

Even on a small in-memory table, the deep page costs about 15 times the cursor. On disk-backed tables with wide rows and real load, deep offsets grow into hundreds of milliseconds or seconds. OFFSET 179980 makes the database walk and discard 179,980 rows on every request, and rows inserted meanwhile shift pages, so users see duplicates or miss items. A keyset cursor seeks straight to “after (created_at, id)” using the index. It costs the same at any depth and is stable while data changes. Always include a unique tie-breaker (the id) in both the ORDER BY and the cursor.

a cursor-paginated endpoint with an opaque cursor

import base64
import json
import sqlite3
from typing import Annotated

from fastapi import FastAPI, HTTPException, Query, Request, Response
from fastapi.testclient import TestClient
from pydantic import BaseModel

db = sqlite3.connect(":memory:", check_same_thread=False)
db.execute("CREATE TABLE products (id INTEGER PRIMARY KEY, name TEXT, price REAL)")
db.executemany("INSERT INTO products (name, price) VALUES (?, ?)", [(f"Product {i:03d}", (i * 37) % 500) for i in range(1, 58)])

def encode(price, pid):
    return base64.urlsafe_b64encode(json.dumps([price, pid]).encode()).decode()

def decode(cursor):
    try:
        price, pid = json.loads(base64.urlsafe_b64decode(cursor))
        return float(price), int(pid)
    except Exception:
        raise HTTPException(400, "invalid cursor") from None

class Product(BaseModel):
    id: int
    name: str
    price: float

class Page(BaseModel):
    items: list[Product]
    next_cursor: str | None

app = FastAPI()

@app.get("/products")
def list_products(request: Request, response: Response,
                  limit: Annotated[int, Query(ge=1, le=50)] = 20, cursor: str | None = None) -> Page:
    after = decode(cursor) if cursor else (-1.0, 0)
    rows = db.execute("SELECT id, name, price FROM products WHERE (price, id) > (?, ?) ORDER BY price, id LIMIT ?",
                      (*after, limit + 1)).fetchall()          # fetch one extra to know if there is a next page
    items = [Product(id=r[0], name=r[1], price=r[2]) for r in rows[:limit]]
    next_cursor = encode(items[-1].price, items[-1].id) if len(rows) > limit else None
    if next_cursor:
        response.headers["Link"] = f'<{request.url.include_query_params(cursor=next_cursor)}>; rel="next"'
    return Page(items=items, next_cursor=next_cursor)

client = TestClient(app)
cursor, pages, seen = None, 0, []
while True:
    r = client.get("/products", params={"limit": 25, **({"cursor": cursor} if cursor else {})}).json()
    pages += 1
    seen += [p["id"] for p in r["items"]]
    cursor = r["next_cursor"]
    if not cursor:
        break
print(f"{pages} pages, {len(seen)} products, all unique: {len(set(seen)) == len(seen)}")
print("tampered cursor:", client.get("/products", params={"cursor": "not-base64!"}).status_code)

Output

3 pages, 57 products, all unique: True
tampered cursor: 400

2. Filtering and sorting without SQL injection

ORDER BY from a whitelist, never from the request string

from typing import Annotated, Literal

from fastapi import FastAPI, Query
from fastapi.testclient import TestClient

SORTS = {"price": "price", "-price": "price DESC", "name": "name", "newest": "created_at DESC"}

app = FastAPI()

@app.get("/search")
def search(sort: Annotated[Literal["price", "-price", "name", "newest"], Query()] = "newest",
           min_price: Annotated[float, Query(ge=0)] = 0):
    order_by = SORTS[sort]                                        # mapped from a closed set: safe
    sql = f"SELECT id, name FROM products WHERE price >= ? ORDER BY {order_by}, id LIMIT 20"
    return {"sql": sql, "params": [min_price]}                    # values ALWAYS as bound parameters

client = TestClient(app)
print(client.get("/search?sort=-price&min_price=100").json())
r = client.get("/search", params={"sort": "price; DROP TABLE products--"})
print(r.status_code, r.json()["detail"][0]["msg"])

Output

{'sql': 'SELECT id, name FROM products WHERE price >= ? ORDER BY price DESC, id LIMIT 20', 'params': [100.0]}
422 Input should be 'price', '-price', 'name' or 'newest'

Bound parameters (?, or SQLAlchemy expressions) protect values. Identifiers (column names, sort directions) cannot be bound, so they must come from a closed mapping. Validate the incoming string with Literal or an Enum, then translate it.

3. Optimistic concurrency with ETags

GET returns an ETag; writes must send If-Match

from typing import Annotated

from fastapi import FastAPI, Header, HTTPException, Response
from fastapi.testclient import TestClient
from pydantic import BaseModel

ORDERS = {1: {"id": 1, "address": "12 MG Road", "version": 3}}

class AddressChange(BaseModel):
    address: str

app = FastAPI()

def etag(order):
    return f'"v{order["version"]}"'

@app.get("/orders/{order_id}")
def get_order(order_id: int, response: Response):
    order = ORDERS[order_id]
    response.headers["ETag"] = etag(order)
    return order

@app.put("/orders/{order_id}/address")
def change_address(order_id: int, body: AddressChange, response: Response,
                   if_match: Annotated[str | None, Header()] = None):
    order = ORDERS[order_id]
    if if_match is None:
        raise HTTPException(428, "send If-Match with the ETag you last read")    # 428 Precondition Required
    if if_match != etag(order):
        raise HTTPException(412, "order changed since you read it; re-read and retry")
    order["address"] = body.address
    order["version"] += 1                  # in SQL: UPDATE … SET version = version + 1 WHERE id = ? AND version = ?
    response.headers["ETag"] = etag(order)
    return order

client = TestClient(app)
tag = client.get("/orders/1").headers["etag"]                     # both users read version 3
asha = client.put("/orders/1/address", json={"address": "Flat 4, Marine Drive"}, headers={"If-Match": tag})
ravi = client.put("/orders/1/address", json={"address": "Old warehouse"}, headers={"If-Match": tag})
print("asha:", asha.status_code, asha.headers.get("etag"), "| ravi:", ravi.status_code, ravi.json()["detail"])
print("no If-Match:", client.put("/orders/1/address", json={"address": "x"}).status_code)

Output

asha: 200 "v4" | ravi: 412 order changed since you read it; re-read and retry
no If-Match: 428

4. Idempotency keys: safe retries for POST

A mobile client sends “create payment”, the network drops, and the client cannot know whether it arrived. If it retries, the customer may be charged twice. With an Idempotency-Key header (a UUID the client generates per operation), the server stores the first result and replays it for repeats:

first call executes, retries replay, key reuse with a different body is rejected

import hashlib
import json
from typing import Annotated

from fastapi import FastAPI, Header, HTTPException, Response
from fastapi.testclient import TestClient
from pydantic import BaseModel

class PaymentIn(BaseModel):
    order_id: int
    amount: int

STORE: dict[str, dict] = {}               # in production: a table with a unique key and an expiry (e.g. 24 h)
CHARGES = []

app = FastAPI()

@app.post("/payments", status_code=201)
def create_payment(body: PaymentIn, response: Response,
                   idempotency_key: Annotated[str, Header(min_length=16, max_length=64)]):
    fingerprint = hashlib.sha256(json.dumps(body.model_dump(), sort_keys=True).encode()).hexdigest()
    if (saved := STORE.get(idempotency_key)) is not None:
        if saved["fingerprint"] != fingerprint:
            raise HTTPException(422, "Idempotency-Key reused with a different request body")
        response.headers["Idempotent-Replayed"] = "true"
        response.status_code = saved["status"]
        return saved["body"]
    CHARGES.append(body.amount)                                   # the side effect happens once
    result = {"payment_id": f"pay_{len(CHARGES)}", "status": "captured", "amount": body.amount}
    STORE[idempotency_key] = {"fingerprint": fingerprint, "status": 201, "body": result}
    return result

client = TestClient(app)
key = {"Idempotency-Key": "2f8c1e0a-7b1d-4c4e-9a55-3c2f1d9e8b71"}
first = client.post("/payments", json={"order_id": 1042, "amount": 49900}, headers=key)
retry = client.post("/payments", json={"order_id": 1042, "amount": 49900}, headers=key)
misuse = client.post("/payments", json={"order_id": 1042, "amount": 1}, headers=key)
print(first.status_code, first.json(), "| retry:", retry.status_code, retry.headers.get("idempotent-replayed"))
print("misuse:", misuse.status_code, "| charges made:", CHARGES)

Output

201 {'payment_id': 'pay_1', 'status': 'captured', 'amount': 49900} | retry: 201 true
misuse: 422 | charges made: [49900]

5. Schema migrations with Alembic

the everyday commands (pip install alembic; not run here)

alembic init -t async migrations                       # async template for SQLAlchemy asyncio engines
alembic revision --autogenerate -m "add orders.version" # diff models vs database → a migration script
alembic upgrade head                                   # apply
alembic downgrade -1                                   # roll back one step (write downgrades!)

Recap