backend / fastapi
Pagination et filtres
Explication
Ce que vous allez apprendre
- Implémenter une pagination offset/limit avec des bornes de sécurité
- Combiner plusieurs filtres dynamiques dans une même requête SQLAlchemy
- Sécuriser un tri dynamique avec une whitelist explicite de colonnes
- Comprendre les avantages de la pagination par curseur sur de gros volumes
- Éviter qu'un client puisse demander une taille de page démesurée
Dans quel contexte ?
Le catalogue de app/routers/products.py compte désormais 2 millions de produits, et l'endpoint GET /products qui renvoyait auparavant toute la table d'un coup fait maintenant timeout systématiquement, saturant à la fois la mémoire du serveur et celle du navigateur qui reçoit la réponse. Ajouter une pagination offset/limit avec une taille de page bornée (page_size max 100) résout le timeout immédiat ; migrer ensuite vers une pagination par curseur évite le décalage de résultats que provoquerait l'ajout constant de nouveaux produits entre deux pages consultées.
Le problème, d'abord
Une table de produits avec des millions de lignes ne peut raisonnablement pas être renvoyée en une seule réponse JSON. Le temps de réponse serait catastrophique, la mémoire du serveur et du client saturée pour rien.
La pagination découpe les résultats en "pages" digestes, renvoyées une par une à la demande du client. Voyons d'abord la méthode la plus intuitive : offset/limit.
Cette approche consiste à demander "saute les N premiers résultats, puis donne-m'en M". Simple à comprendre et à implémenter, elle permet aussi de sauter directement à une page arbitraire.
Sa limite apparaît sur des données qui changent fréquemment entre deux requêtes. Si une ligne est insérée avant la page consultée, un même résultat peut apparaître deux fois, ou un autre être sauté silencieusement.
Une fois la pagination en place, une fonctionnalité pratique s'ajoute souvent : le tri dynamique. Laisser un client choisir la colonne de tri est pratique, mais comporte un vrai risque.
Il ne faut JAMAIS injecter directement cette chaîne fournie par le client dans une requête SQL. Cela ouvrirait la porte à des manipulations dangereuses.
La bonne pratique est une liste blanche explicite de colonnes autorisées. Vérifiée AVANT de construire la requête, jamais après.
Pour les cas où les données changent souvent, il existe une alternative plus robuste : la pagination par curseur. Plutôt que de compter des positions, elle se souvient du DERNIER élément vu.
Elle demande "donne-moi ce qui vient après celui-ci". Cette approche reste stable même si des lignes sont ajoutées ou supprimées entre deux requêtes, et bien plus performante sur de gros volumes.
Ce gain a un prix à connaître : elle ne permet plus de sauter directement à une page arbitraire. Un compromis à évaluer selon le besoin réel de l'interface.
| Stratégie | Stabilité sur données changeantes | Saut à une page arbitraire |
|---|---|---|
| Offset/limit | Faible | Oui |
| Curseur | Élevée | Non |
Piège fréquent
Ne jamais laisser un client demander une taille de page illimitée ou démesurée (page_size=999999), cela annulerait tout l'intérêt de la pagination et pourrait servir de vecteur de déni de service. Toujours borner avec Query(le=100) ou équivalent.
Commandes & code
Pagination et filtres
# Pagination offset/limit — simple, adaptée à la plupart des cas
from fastapi import Query
from typing import Annotated
from pydantic import BaseModel
class PageParams(BaseModel):
page: int = 1
page_size: int = 20
@property
def offset(self) -> int:
return (self.page - 1) * self.page_size
def get_page_params(
page: Annotated[int, Query(ge=1)] = 1,
page_size: Annotated[int, Query(ge=1, le=100)] = 20,
) -> PageParams:
return PageParams(page=page, page_size=page_size)
@router.get("/products")
def list_products(
db: Annotated[Session, Depends(get_db)],
pagination: Annotated[PageParams, Depends(get_page_params)],
):
total = db.query(Product).count()
items = (
db.query(Product)
.offset(pagination.offset)
.limit(pagination.page_size)
.all()
)
return {
"items": items,
"total": total,
"page": pagination.page,
"page_size": pagination.page_size,
"total_pages": (total + pagination.page_size - 1) // pagination.page_size,
}# Filtres multiples combinés dynamiquement
from sqlalchemy import select
@router.get("/products/search")
def search_products(
db: Annotated[Session, Depends(get_db)],
name: str | None = None,
min_price: float | None = None,
max_price: float | None = None,
category: str | None = None,
in_stock: bool | None = None,
):
stmt = select(Product)
if name:
stmt = stmt.where(Product.name.ilike(f"%{name}%"))
if min_price is not None:
stmt = stmt.where(Product.price >= min_price)
if max_price is not None:
stmt = stmt.where(Product.price <= max_price)
if category:
stmt = stmt.where(Product.category == category)
if in_stock is not None:
stmt = stmt.where(Product.stock > 0 if in_stock else Product.stock == 0)
return db.scalars(stmt).all()# Tri dynamique — whitelist stricte des colonnes triables (jamais une string brute dans order_by)
SORTABLE_FIELDS = {"name": Product.name, "price": Product.price, "created_at": Product.created_at}
@router.get("/products/sorted")
def list_sorted_products(
db: Annotated[Session, Depends(get_db)],
sort_by: str = "created_at",
order: str = "desc",
):
if sort_by not in SORTABLE_FIELDS:
raise HTTPException(status_code=400, detail=f"Tri invalide : {sort_by}")
column = SORTABLE_FIELDS[sort_by]
column = column.desc() if order == "desc" else column.asc()
return db.query(Product).order_by(column).all()# Cursor-based pagination — plus stable que offset/limit sur des données qui changent souvent
@router.get("/products/cursor")
def list_products_cursor(
db: Annotated[Session, Depends(get_db)],
cursor: int | None = None, # dernier ID vu par le client
limit: int = 20,
):
stmt = select(Product).order_by(Product.id).limit(limit + 1) # +1 pour savoir s'il y a une page suivante
if cursor is not None:
stmt = stmt.where(Product.id > cursor)
items = list(db.scalars(stmt))
has_next = len(items) > limit
items = items[:limit]
return {
"items": items,
"next_cursor": items[-1].id if has_next else None,
}| Stratégie | Avantage | Inconvénient |
|---|---|---|
| Offset/limit | Simple, permet le "aller à la page N" | Décalage possible si des lignes sont insérées/supprimées entre deux requêtes |
| Cursor-based | Stable même avec des données changeantes, performant sur de gros volumes | Pas de saut direct à une page arbitraire |
Résumé
- La pagination offset/limit couvre la majorité des cas d'usage classiques (admin, listes modérées).
- Toujours borner
page_size(le=100) pour éviter qu'un client ne demande des milliers de lignes d'un coup. - Le tri dynamique doit passer par une whitelist explicite de colonnes, jamais une string brute injectée dans la requête.
- La pagination par curseur (cursor-based) est préférable sur de gros volumes de données fréquemment modifiées.
Exercices pratiques
Mission : le tableau de bord reporting qui affiche deux fois le même produit
Objectif : Diagnostiquer un défaut de pagination offset/limit sur un flux de données changeant, puis sécuriser le tri dynamique et le dimensionnement des pages.
Contexte
L'équipe reporting consulte GET /products avec la pagination offset/limit vue dans cette leçon, pendant qu'un job d'import ajoute en continu de nouveaux produits en base. En passant de la page 2 à la page 3, certains produits apparaissent deux fois et d'autres semblent avoir disparu. Par ailleurs, un vieux lien bookmarké envoie sort_by=internal_notes vers /products/sorted, ce qui fait planter l'endpoint avec une erreur 500 au lieu d'un message clair.