Pular para o conteúdo

    Capítulo 27, Avançado

    Paginação

    Nunca devolva "tudo". Uma lista que cresce sem limite derruba a API, o banco e o navegador do cliente. A paginação entrega os dados em partes, e existem duas formas de fazê-la, com características bem diferentes.

    Paginação por página: `limite` e `pagina`

    A forma mais conhecida: o cliente pede a página N com L itens. O banco pula (N - 1) × L linhas (OFFSET) e devolve L (LIMIT). Há três cuidados: validar os limites (um limite=1000000 seria uma negação de serviço), ordenar sempre (sem ORDER BY, a ordem das linhas é indefinida e as páginas se misturam) e devolver o total, para o cliente saber quantas páginas existem.

    avancado/cap27_paginacao.pylinhas 10 a 86
    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)
    Saída
    95 10 10 {'id': 1, 'nome': 'Produto 001'} 10
5 91
0 422

    A resposta traz tudo o que o cliente precisa: a página atual, o total de registros e de páginas. A última página (a 10) tem apenas 5 itens, e uma página além do fim devolve uma lista vazia (e não um erro). O limite acima de 50 é recusado com 422. O tipo Pagina[ProdutoSaida] é um modelo genérico, que serve para paginar qualquer recurso.

    A busca e o caractere `%`

    O curso original usa ilike(f"%{busca}%"). Funciona, mas o % e o _ têm significado especial no LIKE, e o usuário os controla. Quem busca por % recebe tudo. O icontains(..., autoescape=True) trata esses caracteres como texto comum:

    avancado/cap27_paginacao.pylinhas 91 a 96
    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))
    Saída
    10
0
com ilike e o texto %: 95

    O primeiro número é uma busca real (os produtos 010 a 019). Buscar o texto % devolveu zero na rota segura e devolveu 95 (todos) na versão ingênua.

    O custo escondido do `OFFSET`

    Para entregar a página 5000, o banco precisa ler e descartar as 4999 páginas anteriores. O custo cresce com a profundidade, e há um segundo problema: quando os dados mudam entre duas requisições, a página 2 deixa de ser a continuação da página 1. Se um item da página 1 for apagado, tudo desliza uma posição, e um item é pulado:

    avancado/cap27_paginacao.pylinhas 101 a 107
    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)
    Saída
    [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
[12, 13, 14, 15, 16, 17, 18, 19, 20, 21]

    A segunda página começa no 12: o produto 11 nunca apareceu para o cliente, porque a remoção do 3 empurrou os itens para cima.

    Paginação por cursor: continuar de onde parou

    A alternativa é o cursor (ou keyset): em vez de "pule N", o cliente diz "me dê os itens depois deste". A consulta vira WHERE id > :ultimo ORDER BY id LIMIT :limite, usa o índice da chave primária (custa o mesmo em qualquer profundidade) e não pula nem repete itens quando os dados mudam:

    avancado/cap27_paginacao.pylinhas 112 a 133
    @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)
    Saída
    [11, 12, 13, 14, 15, 16, 17, 18, 19, 20]
94 94 False

    Com o cursor depois=10, a segunda página começa em 11, e o item 11 não foi perdido. Percorrer a lista inteira vê todos os 94 itens restantes (o 3 foi apagado), sem repetir nenhum. Pedir limite + 1 itens é um truque barato para saber se existe uma próxima página, sem uma segunda consulta.

    Página (OFFSET)Cursor (keyset)
    O cliente pede"página 7""depois do item X"
    Pular para uma página qualquerSimNão, só avança
    Custo em páginas fundasCresceConstante
    Dados mudando durante a leituraPula ou repete itensEstável
    Total de páginasFácilEm geral não se informa
    Quando usarTelas com "ir para a página 5", conjuntos pequenosFeeds, rolagem infinita, exportações, tabelas grandes

    Exercício 1

    O total de páginas

    Escreva total_de_paginas(total, limite), que devolva quantas páginas existem (arredondando para cima, e 0 quando não há itens), e confira os casos de borda.