Implementación de Paginación en FastAPI con SQLModel
Resumen
Trabajar con grandes cantidades de datos puede reducir considerablemente el rendimiento de las aplicaciones, ocasionando tiempos de carga lentos. Es aquí donde la paginación cumple un papel fundamental, permitiendo mostrar solo una pequeña parte de los datos disponibles, mejorando así la velocidad de respuesta y optimizando la experiencia del usuario.
¿Qué es la paginación y por qué usarla?
La paginación es una técnica que consiste en dividir grandes conjuntos de datos en segmentos más pequeños, mostrándolos únicamente cuando el usuario así lo requiere. Sus principales ventajas son:
Reducción del tiempo necesario para cargar los datos.
Optimización de recursos al traer información parcial según necesidad.
Mejora notable en la experiencia del usuario al optimizar los tiempos de espera y disponer solo la información solicitada por éste.
¿Cómo crear datos de prueba para paginar?
Para verificar nuestro sistema de paginación, se recomienda generar datos ficticios. El proceso a seguir incluye:
Crear un archivo (por ejemplo: Create Multiple Transactions).
Generar un usuario (Customer) para asociar las transacciones.
Realizar transacciones ficticias mediante un ciclo (loop for), incrementando progresivamente algún valor numérico para diferenciarlas.
Concluir esta creación de datos con un commit en nuestra base de datos.
Este método posibilita realizar pruebas realistas que permitan configurar y verificar la paginación efectivamente.
¿Cómo implementar la paginación usando FastAPI y SQLModel?
Para poner en práctica la paginación en FastAPI usando SQLModel debes seguir estos pasos específicos:
Definir nuevas variables en la URL del endpoint para generar la paginación:
skip:int= Query(0, description="Número de registros a omitir")limit:int= Query(10, description="Número de registros a devolver")
Modificar la consulta (query) realizada a la base de datos para incorporar estas nuevas variables mediante funciones específicas:
Estas modificaciones permitirán que al llamar al endpoint se pueda controlar cuántos y cuáles registros serán mostrados según los parámetros especificados por el usuario, mejorando así el rendimiento de nuestras aplicaciones.
Comparte qué otras aplicaciones consideras útiles para implementar la paginación o qué dudas tienes sobre cómo aplicar esta técnica en tus proyectos.
# Respuesta que incluirá las transacciones y la paginaciónclassPaginatedTransactionsResponse(SQLModel): total_count:int# Total de elementos total_pages:int# Total de páginas current_page:int# Página actual limit:int# Límite de elementos por página transactions: List["Transaction"]# Transacciones en la página actual```**Modificar router:**
```python
@router.get("/transactions", response_model=PaginatedTransactionsResponse)asyncdeflist_transactions( session: SessionDep, skip:int= Query(0, description="Registros a omitir"), limit:int= Query(10, description="Número de registros por página"),):# Consulta de transacciones con paginación query = select(Transaction).offset(skip).limit(limit) transactions = session.exec(query).all()# Obtener el total de registros en la base de datos (sin paginación) total_count_query = select(Transaction) total_count =len(session.exec(total_count_query).all())# Contamos los registros# Calcular el total de páginas total_pages =(total_count + limit -1)// limit # Redondear hacia arriba# Crear la respuesta paginada response = PaginatedTransactionsResponse( total_count=total_count, total_pages=total_pages, current_page=(skip // limit)+1,# Calcular la página actual limit=limit, transactions=transactions
)return response
Gracias
Muy buena solucion Bro, yo me ahorraria el modelo, pero muy buena solucion. Saludos!
Como borramos y creamos la base de datos, con IA cree un seed_data para no estar batallando con las pruebas se las comparto
from models import Customer, Plan, CustomerPlan, Transaction, StatusEnum
# -----------------------------
# Datos base
# -----------------------------
names = [
"Carlos", "Ana", "Luis", "Maria", "Pedro",
"Sofia", "Jorge", "Lucia", "Miguel", "Elena"
]
descriptions = [
"Compra en tienda",
"Pago de servicio",
"Transferencia",
"Suscripción",
"Pago en línea",
"Compra de producto",
"Recarga",
"Pago mensual"
]
plan_names = [
("Basic", 100),
("Standard", 200),
("Premium", 300),
("VIP", 500)
]
# -----------------------------
# Seed function
# -----------------------------
def seed():
with Session(engine) as session:
print("🚀 Iniciando seed...")
# -----------------------------
# 1. Crear Customers
# -----------------------------
customers = []
for i in range(10):
customer = Customer(
name=names[i],
email=f"user{i}@test.com",
age=random.randint(18, 60),
description="Usuario generado automáticamente"
)
customers.append(customer)
session.add_all(customers)
session.commit()
for c in customers:
session.refresh(c)
print("✅ Customers creados")
# -----------------------------
# 2. Crear Plans
# -----------------------------
plans = []
for name, price in plan_names:
plan = Plan(
name=name,
price=price,
description=f"Plan {name}"
)
plans.append(plan)
session.add_all(plans)
session.commit()
for p in plans:
session.refresh(p)
print("✅ Plans creados")
# -----------------------------
# 3. Asignar planes a customers
# -----------------------------
customer_plans = []
for customer in customers:
num_plans = random.choice([1, 2])
selected_plans = random.sample(plans, num_plans)
if num_plans == 1:
cp = CustomerPlan(
customer_id=customer.id,
plan_id=selected_plans[0].id,
status=StatusEnum.ACTIVE
)
customer_plans.append(cp)
else:
cp_active = CustomerPlan(
customer_id=customer.id,
plan_id=selected_plans[0].id,
status=StatusEnum.ACTIVE
)
cp_inactive = CustomerPlan(
customer_id=customer.id,
plan_id=selected_plans[1].id,
status=StatusEnum.INACTIVE
)
customer_plans.extend([cp_active, cp_inactive])
session.add_all(customer_plans)
session.commit()
print("✅ Planes asignados a customers")
# -----------------------------
# 4. Crear Transactions
# -----------------------------
transactions = []
for _ in range(100):
transaction = Transaction(
amount=random.randint(10, 1000),
description=random.choice(descriptions),
customer_id=random.choice(customers).id
)
transactions.append(transaction)
session.add_all(transactions)
session.commit()
print("✅ 100 transacciones creadas")
print("🎉 Seed completado con éxito")
# -----------------------------
# Ejecutar
# -----------------------------
if __name__ == "__main__":
seed()
Mi solución al reto es la siguiente, encontré "func" para ejecutar el COUNT(*) esto para el total de items, en cuanto a los otros datos de páginación todos los podemos calcular con simple matemática:
Se debe usar una Query para los parámetros skip y limit en FastAPI por varias razones:
Flexibilidad: Permite que el usuario especifique la cantidad de elementos a omitir y el número máximo a devolver directamente en la URL, facilitando la personalización de las consultas.
Documentación Automática: FastAPI genera automáticamente documentación interactiva que incluye estos parámetros como parte de la API, mejorando la experiencia del desarrollador.
Control de Errores: Se pueden agregar validaciones fácilmente, como restringir valores negativos o establecer límites máximos, lo que no sería tan sencillo con un entero fijo.
Esto optimiza el manejo de grandes volúmenes de datos y mejora la experiencia del usuario.
@router.get("/transactions", tags=["transaction"])asyncdeflist_transaction( session: SessionDep, skip:int= Query(0, description="Registros a omitir"), limit:int= Query(10, description="Cantidad de registros a retornar"),): statement = select(func.count()).select_from(Transaction) number_of_transactions = session.exec(statement).all()[0]if skip >= number_of_transactions:raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="El valor de skip excede el número de transacciones",) number_pages =(number_of_transactions + limit -1)// limit
if(skip // limit)+1> number_pages:raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="La página no existe",) actual_page =(skip // limit)+1 query = select(Transaction).offset(skip).limit(limit) transactions = session.exec(query).all()return{"total_transactions": number_of_transactions,"number_pages": number_pages,"actual_page": actual_page,"transactions": transactions,}
Me causaba un poco de ruido usar len() asi que busque una forma de ejecutar una funcion count directamente en el SQL.
Está bien que no usemos len
FastApi da la solución a la paginación con pocas lineas de código. Aquí les dejo la documentación:
Aqui les dejo mi solucion:
Primero se instala la libreria requerida: ```js
pip install fastapi-pagination
En main.py agregamos la importacion de la libreria e inicializamos:```js
from fastapi_pagination import add_pagination
app = FastAPI(lifespan=create_all_tables)
// Incluir routers
// Inicializamos la paginacion de
// la API para poder usarse en toda
// la aplicacion
add_pagination(app)
En el archivo router de transactions.py realizamos lo siguiente:```js
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
FastApi da la solución a la paginación con pocas lineas de código. Aquí les dejo la documentación:
Aqui les dejo mi solucion:
Primero se instala la libreria requerida:
pip install fastapi-pagination
````pip install fastapi-pagination`* En main.py agregamos la importacion de la libreria e inicializamos:
```js
from fastapi_pagination import add_pagination
app =FastAPI(lifespan=create_all_tables)// Incluir routers // Inicializamos la paginacion de // la API para poder usarse en toda // la aplicacion add_pagination(app)// Incluir routers // Inicializamos la paginacion de // la API para poder usarse en toda // la aplicacionadd_pagination(app)
En el archivo router de transactions.py realizamos lo siguiente:
el simbolo -> en realidad es "->" sin las comillas
La paginación es esencial para manejar grandes cantidades de datos en las API, mejorando el rendimiento y la experiencia del usuario. FastAPI facilita la implementación de paginación combinando parámetros de consulta (query parameters) y modelos de datos.
A continuación, te explico cómo implementarla paso a paso:
1. Definición del Modelo de Datos
Usaremos un modelo para representar los datos que queremos paginar.
from typing import List
from pydantic import BaseModel
class Item(BaseModel):
id: int
name: str
description: str
2. Creación de Datos Simulados
Generemos algunos datos simulados para demostrar la paginación.
items = [
{"id": i, "name": f"Item {i}", "description": f"Description for item {i}"}
for i in range(1, 101)
]
3. Configuración de Parámetros de Paginación
Definimos parámetros de consulta (query parameters) como skip (elementos a omitir) y limit (número de elementos a devolver).
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/", response_model=List[Item])
async def get_items(skip: int = Query(0, ge=0), limit: int = Query(10, ge=1)):
return items[skip : skip + limit]
4. Uso del Endpoint
Solicitud: GET /items/?skip=10&limit=5
Respuesta: Los 5 elementos que comienzan desde el índice 10.
Valores predeterminados razonables: Define un limit por defecto para evitar sobrecargar el servidor.
Máximo permitido: Establece un valor máximo para limit si necesitas evitar cargas excesivas.limit: int = Query(10, ge=1, le=100)
Filtros adicionales: Agrega parámetros de consulta para buscar o filtrar los datos.
Conclusión
La implementación de paginación en FastAPI es directa, ya sea con datos simulados o bases de datos. Aprovecha los modelos de Pydantic para estructurar las respuestas y las consultas SQL para optimizar el rendimiento en bases de datos.
EN MODELS.PYclass <u>CustomerPage</u>(<u>BaseModel</u>):
El reto, lo hice diferente, puedo escoger cuantos datos extraigo por pagina, seleccionar el numero de pagina que quiero visualizar, y ver el numero total de paginas (segun la cantidad de datos por pagina).
@router.get("/transactions/number", tags=["transactions"])asyncdeflist_number_transactions( session: SessionDep, registros_por_pagina:int= Query(10, description="Número de registros por pagina"), numero_pagina:int= Query(1, description="Número de página"),)->list[Transaction]:'''
Retorna una lista de todos los clientes en la base de datos.
* Parámetros:
- session: La sesión de base de datos.
* Retorna:
- Una lista de clientes
''' transaction_db:int= session.exec(select(Transaction.id)).all() number_transaction :int=len(transaction_db) number_pages = math.ceil(number_transaction / registros_por_pagina) skip =((numero_pagina -1)* registros_por_pagina) query = select(Transaction).offset(skip).limit(registros_por_pagina) transactions = session.exec(query).all() count_pages:str=f"Cantidad de páginas: {number_pages}"return transactions, count_pages
Lo hice de una manera mas sencilla, retornando un diccionario:
@router.get("/transactions")asyncdeflist_transaction(session:SessionDep, skip:int= Query(0, description="Registros a omitir"), limit:int= Query(10, description="numero de registro")): query = select(Transaction).offset(skip).limit(limit) amount =len(session.exec(select(Transaction)).all())return{"pages":(amount-skip)/(limit-skip),"amount_per_page":limit-skip,"elements":session.exec(query).all()}
Mi solucion> Es importante ubicar las transacciones realizadas by customer...
classPaginatedTransactions(SQLModel):per_page: int =Field(default=10)total_pages: int =Field(default=0)current_page: int =Field(default=1)customer_id: int =Field(foreign_key="customer.id", nullable=False)transactions:List[Transaction]=Field(default_factory=list)``````js
@router.get("/transactions", response_model=PaginatedTransactions, tags=["Transactions"])async def list_transactions(session:SessionDep,customer_id: int =Query(..., description="Customer ID"),skip: int =Query(0, description="Skip register"),limit: int =Query(4, description="Limit register")): customer = session.get(Customer, customer_id)if not customer: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Customer not found!") total_transactions = session.exec(select(Transaction).where(Transaction.customer_id== customer_id)).all() # Get all transactions for the customer
total_count =len(total_transactions) # Get the total number of transactions for the customer
query =select(Transaction).where(Transaction.customer_id== customer_id).offset(skip).limit(limit) transactions = session.exec(query).all()if not transactions: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No transactions found!") total_pages =(total_count + limit -1)// limit # Calculate total pages current_page =(skip // limit) + 1 response =PaginatedTransactions( per_page=limit, total_pages=total_pages, current_page=current_page, customer_id=customer_id, transactions=transactions,)return response
la paginacion....
Excelente curso!
models.py
Genere una clase nueva para el response con la paginación
Se agrega el total de paginas y la pagina actual a la respuesta endpoint:class ResponseListTransactions(BaseModel): transactions: list[Transaction] message: dict
