Paginación y Filtros en FastAPI con SQLite

¿Qué es la paginación?

Cuando una tabla tiene muchos registros (ej: miles de productos), no conviene devolverlos todos de una sola vez porque:

  • La respuesta será muy grande y lenta.
  • Consumirá muchos recursos del servidor y del cliente.
  • La mayoría de las veces el usuario solo necesita ver una parte de los datos.

👉 La paginación permite dividir los resultados en páginas, devolviendo una cantidad limitada (limit) y desde una posición determinada (offset).

Ejemplo:

  • Página 1 — limit=5&offset=0
  • Página 2 — limit=5&offset=5
  • Página 3 — limit=5&offset=10

🔹 ¿Qué son los filtros?

Los filtros permiten devolver solo los registros que cumplen ciertas condiciones.

Ejemplos de filtros:

  • Buscar productos que contengan la palabra "cuaderno".
  • Filtrar por precio mínimo (precio_min=1000).
  • Filtrar por precio máximo (precio_max=3000).
  • Combinar filtros y paginación: q=lapicera&precio_max=2000&limit=3&offset=0.

👉 Así el usuario recibe exactamente los datos que necesita y no todo el catálogo.

📂 Estructura del proyecto

.
 main.py         Archivo principal de la aplicación FastAPI
 database.py     Configuración de la base de datos y sesión
 models.py       Modelo ORM de la tabla productos
 schemas.py      Validación y serialización con Pydantic
 crud.py         Lógica para acceder a la base con filtros y paginación

database.py — Conexión y sesión

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase):
    pass

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

models.py — Tabla productos

from sqlalchemy import Column, Integer, String, Float
from database import Base

class Producto(Base):
    __tablename__ = "productos"

    codigo = Column(Integer, primary_key=True, index=True, autoincrement=True)
    descripcion = Column(String(100), nullable=False)
    precio = Column(Float, nullable=False)

schemas.py — Modelos de salida

from pydantic import BaseModel

class ProductoOut(BaseModel):
    codigo: int
    descripcion: str
    precio: float

    class Config:
        from_attributes = True

class ProductosPage(BaseModel):
    total: int
    limit: int
    offset: int
    items: list[ProductoOut]

crud.py — Consulta con filtros y paginación

from typing import Optional, Tuple, List
from sqlalchemy.orm import Session
from sqlalchemy import func
from models import Producto

# Campos permitidos para ordenar
_ORDERABLE_FIELDS = {
    "codigo": Producto.codigo,
    "descripcion": Producto.descripcion,
    "precio": Producto.precio,
}

def listar_productos(
    db: Session,
    q: Optional[str] = None,
    precio_min: Optional[float] = None,
    precio_max: Optional[float] = None,
    ordenar_por: str = "codigo",
    orden: str = "asc",
    limit: int = 10,
    offset: int = 0,
) -> Tuple[int, List[Producto]]:
    query = db.query(Producto)

    # Filtros
    if q:
        query = query.filter(Producto.descripcion.ilike(f"%{q}%"))
    if precio_min is not None:
        query = query.filter(Producto.precio >= precio_min)
    if precio_max is not None:
        query = query.filter(Producto.precio <= precio_max)

    # Total antes de paginar
    total = query.with_entities(func.count()).scalar() or 0

    # Orden
    col = _ORDERABLE_FIELDS.get(ordenar_por, Producto.codigo)
    col = col.desc() if orden.lower() == "desc" else col.asc()

    # Paginación
    items = query.order_by(col).offset(offset).limit(limit).all()

    return total, items

main.py — Semilla + endpoint /productos

from fastapi import FastAPI, Depends, Query
from sqlalchemy.orm import Session
from database import Base, engine, get_db, SessionLocal
from models import Producto
from schemas import ProductosPage
from contextlib import asynccontextmanager
import crud

# Crear tablas
Base.metadata.create_all(bind=engine)

# Sembrar 10 productos al iniciar
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 🌱 Sembrar datos al iniciar
    db = SessionLocal()
    try:
        if db.query(Producto).count() == 0:
            db.add_all([
                Producto(descripcion="Lapicera azul",        precio=1200.0),
                Producto(descripcion="Cuaderno A4",          precio=3500.0),
                Producto(descripcion="Regla 30 cm",          precio=1800.0),
                Producto(descripcion="Marcador indeleble",   precio=2200.0),
                Producto(descripcion="Lápiz HB",             precio=800.0),
                Producto(descripcion="Goma de borrar",       precio=700.0),
                Producto(descripcion="Carpeta anillada",     precio=4200.0),
                Producto(descripcion="Resaltador amarillo",  precio=1500.0),
                Producto(descripcion="Tijera escolar",       precio=2600.0),
                Producto(descripcion="Pegamento escolar",    precio=1300.0),
            ])
            db.commit()
    finally:
        db.close()
    yield   # 👈 Aquí arranca la aplicación


app = FastAPI(title="FastAPI + SQLite — Paginación y Filtros", version="1.0.0",lifespan=lifespan)

@app.get("/productos", response_model=ProductosPage)
def listar_productos(
    db: Session = Depends(get_db),
    q: str | None = Query(None, description="Buscar por descripción"),
    precio_min: float | None = Query(None, ge=0),
    precio_max: float | None = Query(None, ge=0),
    ordenar_por: str = Query("codigo", pattern="^(codigo|descripcion|precio)$"),
    orden: str = Query("asc", pattern="^(asc|desc)$"),
    limit: int = Query(10, ge=1, le=100),
    offset: int = Query(0, ge=0),
):
    total, items = crud.listar_productos(
        db=db,
        q=q,
        precio_min=precio_min,
        precio_max=precio_max,
        ordenar_por=ordenar_por,
        orden=orden,
        limit=limit,
        offset=offset,
    )
    return {
        "total": total,
        "limit": limit,
        "offset": offset,
        "items": items,
    }

Se cargan, la primera vez que ejecutamos la aplicación 10 filas en la tabla productos.

🚀 Cómo probar

Ejecutar el servidor:

uvicorn main:app --reload

Probar en Swagger UI:

👉 http://127.0.0.1:8000/docs

Ejemplos de uso

Paginación:
/productos?limit=5&offset=5

Búsqueda por texto:
/productos?q=cuaderno

Rango de precios:
/productos?precio_min=1000&precio_max=2500

Orden descendente por precio:
/productos?ordenar_por=precio&orden=desc

Podemos probar los ejemplos con la documentación automática http://127.0.0.1:8000/docs

documentacion docs
Retornar