Chapter 27, Advanced
Pagination
Never return "everything". A list that grows without limit crashes the API, the database, and the client's browser. Pagination delivers the data in parts, and there are two ways to do it, with very different characteristics.
Pagination by page: `limit` and `page`
The most well-known way: the client requests page N with L items. The database skips (N - 1) × L rows (OFFSET) and returns L (LIMIT). There are three precautions: validate the limits (a limite=1000000 would be a denial of service), always sort (without ORDER BY, the order of the rows is undefined and the pages mix) and return the total, so the client knows how many pages exist.
from math import ceil
from typing import Annotated, Generic, TypeVar
from fastapi import Depends, FastAPI, Query
from fastapi.testclient import TestClient
from pydantic import BaseModel, ConfigDict
from sqlalchemy import String, create_engine, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker
from sqlalchemy.pool import StaticPool
class Base(DeclarativeBase):
pass
class Produto(Base):
__tablename__ = "produtos"
id: Mapped[int] = mapped_column(primary_key=True)
nome: Mapped[str] = mapped_column(String(50))
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(engine, expire_on_commit=False)
with SessionLocal() as sessao:
sessao.add_all([Produto(nome=f"Produto {i:03d}") for i in range(1, 96)])
sessao.commit()
T = TypeVar("T")
class Pagina(BaseModel, Generic[T]):
pagina: int
limite: int
total: int
paginas: int
dados: list[T]
class ProdutoSaida(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
nome: str
def obter_sessao():
with SessionLocal() as sessao:
yield sessao
Sessao = Annotated[Session, Depends(obter_sessao)]
app = FastAPI()
@app.get("/produtos", response_model=Pagina[ProdutoSaida])
def listar(
sessao: Sessao,
pagina: Annotated[int, Query(ge=1)] = 1,
limite: Annotated[int, Query(ge=1, le=50)] = 10,
busca: str | None = None,
):
consulta = select(Produto).order_by(Produto.id)
if busca:
consulta = consulta.where(Produto.nome.icontains(busca, autoescape=True))
total = sessao.scalar(select(func.count()).select_from(consulta.subquery()))
itens = sessao.scalars(consulta.offset((pagina - 1) * limite).limit(limite)).all()
return {"pagina": pagina, "limite": limite, "total": total, "paginas": ceil(total / limite), "dados": itens}
cliente = TestClient(app)
p1 = cliente.get("/produtos").json()
print(p1["total"], p1["paginas"], len(p1["dados"]), p1["dados"][0], p1["dados"][-1]["id"])
ultima = cliente.get("/produtos?pagina=10").json()
print(len(ultima["dados"]), ultima["dados"][0]["id"])
print(len(cliente.get("/produtos?pagina=99").json()["dados"]), cliente.get("/produtos?limite=51").status_code)
95 10 10 {'id': 1, 'nome': 'Produto 001'} 10
5 91
0 422
The response brings everything the client needs: the current page, the total number of records, and the total number of pages. The last page (page 10) has only 5 items, and a page beyond the end returns an empty list (and not an error). The limite above 50 is rejected with 422. The type Pagina[ProdutoSaida] is a genericmodel, which serves to paginate any resource.
The search and the character `%`
The original course uses ilike(f"%{busca}%"). It works, but the % and the _ have special meaning in the LIKE, and the user controls them. Those searching for % receive everything. The icontains(..., autoescape=True) treats these characters as plain text:
print(cliente.get("/produtos?busca=to 01").json()["total"])
print(cliente.get("/produtos?busca=%25").json()["total"])
with SessionLocal() as sessao:
ingenuo = sessao.scalars(select(Produto).where(Produto.nome.ilike("%" + "%" + "%"))).all()
print("com ilike e o texto %:", len(ingenuo))
10
0
com ilike e o texto %: 95
The first number is a real search (products 010 to 019). Searching for the text % returned zero on the safe route and returned 95 (all) on the naive version.
The hidden cost of `OFFSET`
To deliver page 5000, the database needs to read and discard the 4999 previous pages. The cost grows with depth, and there is a second problem: when the data changes between two requests, page 2 ceases to be a continuation of page 1. If an item from page 1 is deleted, everything shifts one position, and an item is skipped:
with SessionLocal() as sessao:
primeira = [p.id for p in sessao.scalars(select(Produto).order_by(Produto.id).limit(10))]
sessao.delete(sessao.get(Produto, 3))
sessao.commit()
segunda = [p.id for p in sessao.scalars(select(Produto).order_by(Produto.id).offset(10).limit(10))]
print(primeira)
print(segunda)
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
[12, 13, 14, 15, 16, 17, 18, 19, 20, 21]
The second page starts at 12: product 11 never appeared to the customer because the removal of 3 pushed the items up.
Cursor pagination: continue from where you left off
The alternative is the cursor (or keyset): instead of "skip N", the client says "give me the items after this". The query becomes WHERE id > :ultimo ORDER BY id LIMIT :limite, uses the primary key index (costs the same at any depth) and does not skip or repeat items when the data changes:
@app.get("/produtos/cursor")
def listar_por_cursor(
sessao: Sessao,
depois: int = 0,
limite: Annotated[int, Query(ge=1, le=50)] = 10,
):
itens = sessao.scalars(
select(Produto).where(Produto.id > depois).order_by(Produto.id).limit(limite + 1)
).all()
tem_mais = len(itens) > limite
itens = itens[:limite]
return {"dados": [{"id": p.id} for p in itens], "proximo": itens[-1].id if tem_mais else None}
print([p["id"] for p in cliente.get("/produtos/cursor?depois=10").json()["dados"]])
vistos, cursor = [], 0
while cursor is not None:
pagina = cliente.get(f"/produtos/cursor?depois={cursor}&limite=20").json()
vistos += [p["id"] for p in pagina["dados"]]
cursor = pagina["proximo"]
print(len(vistos), len(set(vistos)), 3 in vistos)
[11, 12, 13, 14, 15, 16, 17, 18, 19, 20]
94 94 False
With the cursor depois=10, the second page starts at 11, and item 11 was not lost. Traversing the entire list sees all 94 remaining items (item 3 was deleted), without repeating any. Asking for limite + 1 items is a cheap trick to know if there is a next page, without a second query.
Page (OFFSET) | Cursor (keyset) | |
|---|---|---|
| The client requests | "page 7" | "after item X" |
| Jump to any page | Yes | No, it just advances |
| Cost in deep pages | Grows | Constant |
| Data changing during reading | Skips or repeats items | Stable |
| Total pages | Easy | Generally not reported |
| When to use | Screens with "go to page 5", small sets | Feeds, infinite scrolling, exports, large tables |
Exercise 1
The total number of pages
Write total_de_paginas(total, limite), which returns how many pages there are (rounding up, and 0 when there are no items), and check the edge cases.