Paso 3: Filtros, Búsqueda y Paginación

Descripción

Las APIs reales casi nunca devuelven listas completas. Una API de tareas con 10 000 registros no puede enviar todo de golpe: saturaría la red, ralentizaría el cliente y consumiría memoria innecesariamente. Por eso, los endpoints de listado en producción combinan filtros (para acotar qué se devuelve), búsqueda (para encontrar texto), ordenamiento (para controlar el criterio) y paginación (para limitar cantidad por petición).

En esta cápsula modifies el endpoint GET /tasks para soportar filtros por status y priority, búsqueda por texto en el título, ordenamiento por fecha o prioridad, y paginación con skip y limit. La respuesta incluirá total, skip, limit e items para que el cliente sepa cuántos resultados hay y en qué página está. Es el mismo patrón que usan GitHub, Stripe o cualquier API REST seria.

---

Objetivos de esta cápsula

• 📌 Filtrar por status (pending, in_progress, completed)
• 📌 Filtrar por priority (low, medium, high)
• 🔍 Buscar por texto en title (case insensitive)
• 🔀 Ordenar por created_at o priority
• 📄 Paginar con skip y limit

---

El Patrón

Casi todas las APIs de listado siguen el mismo flujo mental:

1. Filtrar — reducir el conjunto según criterios (status, priority, fechas, etc.)
2. Ordenar — aplicar un criterio de orden (fecha, prioridad, nombre)
3. Paginar — tomar solo un segmento (skip, limit) del resultado
4. Devolver con metadata — incluir total, skip, limit para que el cliente construya la UI (páginas, "mostrando X de Y", etc.)

Puedes pensarlo como un pipeline:

[dataset completo] → FILTRO → ORDEN → PAGINAR → [items + total, skip, limit]

Este patrón lo usan, entre otros:

• GitHub API: GET /repos/{owner}/{repo}/issues?state=open&per_page=30&page=1
• Stripe API: GET /v1/customers?limit=10&created[gte]=...
• Slack API: GET /conversations.list?limit=100&cursor=...

En tu caso usarás skip/limit en lugar de page/per_page (verás el porqué más abajo). El resultado es el mismo: el cliente recibe una porción controlada de datos más metadata para saber cómo pedir la siguiente.

---

Especificación de query parameters

Parámetro	Tipo	Default	Descripción
status	str | None	None	Filtrar por estado exacto
priority	str | None	None	Filtrar por prioridad exacta
search	str | None	None	Buscar texto en title (contains, case insensitive)
sort_by	str	"created_at"	Campo para ordenar: "created_at", "priority", "title"
order	str	"desc"	"asc" o "desc"
skip	int	0	Offset para paginación
limit	int	10	Máximo items por página

---

Formato de respuesta

{
  "total": 15,
  "skip": 0,
  "limit": 10,
  "items": [
    {
      "id": 1,
      "title": "...",
      "description": "...",
      "status": "completed",
      "priority": "high",
      "created_at": "2025-03-01T10:00:00"
    }
  ]
}

---

Modelo para la respuesta paginada

class TaskListResponse(BaseModel):
    total: int
    skip: int
    limit: int
    items: list[TaskResponse]

---

Lógica de filtrado

Orden de aplicación:

1. Filtro status — Si se envía, mantener solo tareas con ese status
2. Filtro priority — Si se envía, mantener solo tareas con esa priority
3. Búsqueda — Si se envía search, mantener solo tareas cuyo title contenga el texto (case insensitive)
4. Ordenamiento — Ordenar por sort_by en orden asc o desc
5. Paginación — Aplicar skip y limit al resultado

---

Por qué filtrado con Optional y default=None

Los parámetros de filtro usan Optional[str] = None (o Query(None, ...)) porque el cliente no siempre quiere filtrar. Si usaras status: str = "pending", cada llamada a GET /tasks filtraría por "pending" aunque no enviaras el parámetro.

• Sin parámetro → status is None → no aplicas filtro → devuelves todas las tareas
• Con ?status=pending → status == "pending" → aplicas filtro

Así el endpoint se comporta como "lista todo" cuando no hay filtros y "lista lo que cumple X" cuando los hay.

---

Por qué búsqueda case insensitive y solo en title

Buscamos con .lower() en ambos lados porque "Proyecto", "proyecto" y "PROYECTO" deben coincidir. La comparación directa "proyecto" in "Proyecto" falla; "proyecto" in "proyecto".lower() funciona.

Solo buscamos en title por simplicidad y porque suele ser el campo más corto y relevante. Buscar en description implicaría más texto y quizá más ruido. En una versión posterior podrías añadir un parámetro search_in o permitir búsqueda en varios campos.

---

Por qué PRIORITY_ORDER para ordenar por prioridad

Ordenar strings alfabéticamente da un resultado incorrecto:

• "high" < "low" alfabéticamente → "high" iría después de "low"
• Lo que quieres es: high > medium > low (por importancia)

Por eso usas un diccionario de prioridad numérica:

PRIORITY_ORDER = {"high": 0, "medium": 1, "low": 2}

def priority_sort_key(task):
    return PRIORITY_ORDER.get(task["priority"], 3)

Ordenar por priority_sort_key da: high primero, luego medium, luego low. El valor por defecto 3 hace que prioridades desconocidas queden al final.

---

Ordenamiento por prioridad

Si sort_by es "priority", necesitas un orden definido. Puedes mapear:

PRIORITY_ORDER = {"high": 0, "medium": 1, "low": 2}

def priority_sort_key(task):
    return PRIORITY_ORDER.get(task["priority"], 3)

Ordenar por priority_sort_key da: high primero, luego medium, luego low.

---

Por qué paginación con skip/limit y no page/per_page

Con page y per_page:

• página 1 → offset 0
• página 2 → offset 10
• fórmula: offset = (page - 1) * per_page

Con skip y limit:

• offset directo: skip=0, skip=10, skip=20
• no hay que calcular página; el cliente ya sabe desde qué índice empieza
• facilita "infinite scroll" y respuestas con cursor

Usar skip/limit es común en MongoDB, PostgreSQL con OFFSET/LIMIT y muchas APIs. Si prefieres page para una UI clásica, puedes derivarlo: page = (skip // limit) + 1.

---

Implementación del endpoint

from typing import Optional

PRIORITY_ORDER = {"high": 0, "medium": 1, "low": 2}


class TaskListResponse(BaseModel):
    total: int
    skip: int
    limit: int
    items: list[TaskResponse]


@app.get("/tasks", response_model=TaskListResponse)
def list_tasks(
    status: Optional[str] = None,
    priority: Optional[str] = None,
    search: Optional[str] = None,
    sort_by: str = "created_at",
    order: str = "desc",
    skip: int = 0,
    limit: int = 10,
):
    result = list(tasks_db)

    # Filtro por status
    if status is not None:
        result = [t for t in result if t.get("status") == status]

    # Filtro por priority
    if priority is not None:
        result = [t for t in result if t.get("priority") == priority]

    # Búsqueda en title
    if search is not None and search.strip():
        search_lower = search.lower().strip()
        result = [t for t in result if search_lower in t.get("title", "").lower()]

    total = len(result)

    # Ordenamiento
    if sort_by == "created_at":
        result.sort(key=lambda t: t.get("created_at", ""), reverse=(order.lower() == "desc"))
    elif sort_by == "priority":
        result.sort(
            key=lambda t: PRIORITY_ORDER.get(t.get("priority", ""), 3),
            reverse=(order.lower() == "desc"),
        )
    elif sort_by == "title":
        result.sort(key=lambda t: t.get("title", "").lower(), reverse=(order.lower() == "desc"))

    # Paginación
    paginated = result[skip : skip + limit]

    return TaskListResponse(total=total, skip=skip, limit=limit, items=paginated)

---

Prueba en /docs — Filtros

1. Filtro por status

   GET /tasks?status=pending

   Esperado: solo tareas con status=pending. En los datos de ejemplo, la tarea "Agregar filtros y paginación".

2. Filtros combinados

   GET /tasks?status=pending&priority=high

   Esperado: tareas que cumplan ambos criterios. Si no hay ninguna, items: [] y total: 0.

3. Sin filtros

   GET /tasks

   Esperado: todas las tareas (las 3 de ejemplo).

---

Prueba en /docs — Búsqueda

4. Búsqueda por texto

   GET /tasks?search=proyecto

   Esperado: tareas cuyo título contenga "proyecto" (p. ej. "Configurar proyecto FastAPI"). La búsqueda es insensible a mayúsculas: search=Proyecto da el mismo resultado.

5. Búsqueda inexistente

   GET /tasks?search=xyz123

   Esperado: items: [], total: 0.

---

Prueba en /docs — Ordenamiento

6. Orden por prioridad ascendente

   GET /tasks?sort_by=priority&order=asc

   Esperado: high primero, luego medium, luego low (gracias a PRIORITY_ORDER).

7. Orden por fecha descendente

   GET /tasks?sort_by=created_at&order=desc

   Esperado: la más reciente primero (la del 2025-03-03).

---

Prueba en /docs — Paginación

8. Primera página (2 items)

   GET /tasks?skip=0&limit=2

   Esperado: items con 2 tareas, total: 3, skip: 0, limit: 2.

9. Segunda página

   GET /tasks?skip=2&limit=2

   Esperado: 1 tarea restante, total: 3, skip: 2, limit: 2.

---

Prueba en /docs — Combinando todo

10. Filtro + búsqueda + orden + paginación

    GET /tasks?status=pending&search=proyecto&sort_by=created_at&order=desc&skip=0&limit=5

    Aquí aplicas todos los parámetros a la vez. El orden interno del pipeline es: primero filtros y búsqueda, luego orden, al final paginación. La respuesta sigue el formato con total, skip, limit e items.

11. Cómo usar total en el frontend

    Con total: 15 y limit: 10 puedes calcular ceil(15/10) = 2 páginas. La siguiente página sería ?skip=10&limit=10. Un botón "Siguiente" enviaría skip = skip_actual + limit.

---

Validación de parámetros (opcional)

Para evitar valores inválidos en sort_by y order, puedes usar Query con Literal:

from fastapi import Query

sort_by: Literal["created_at", "priority", "title"] = Query(default="created_at")
order: Literal["asc", "desc"] = Query(default="desc")

Si el cliente envía sort_by=invalid, FastAPI retornará 422 automáticamente.

---

Código completo — app/main.py (actualizado)

from datetime import datetime
from typing import Literal, Optional

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field


# --- Modelos Pydantic ---

StatusType = Literal["pending", "in_progress", "completed"]
PriorityType = Literal["low", "medium", "high"]


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200, description="Título de la tarea")
    description: str = Field(default="", max_length=1000, description="Descripción opcional")
    status: StatusType = Field(default="pending", description="Estado de la tarea")
    priority: PriorityType = Field(default="medium", description="Prioridad")


class TaskUpdate(BaseModel):
    title: Optional[str] = Field(default=None, min_length=1, max_length=200)
    description: Optional[str] = Field(default=None, max_length=1000)
    status: Optional[StatusType] = None
    priority: Optional[PriorityType] = None


class TaskResponse(BaseModel):
    id: int
    title: str
    description: str
    status: str
    priority: str
    created_at: datetime

    model_config = {"from_attributes": True}


class TaskListResponse(BaseModel):
    total: int
    skip: int
    limit: int
    items: list[TaskResponse]


# --- Base de datos y utilidades ---

PRIORITY_ORDER = {"high": 0, "medium": 1, "low": 2}


def _parse_dt(s: str) -> datetime:
    return datetime.fromisoformat(s.replace("Z", "+00:00"))


tasks_db: list[dict] = [
    {
        "id": 1,
        "title": "Configurar proyecto FastAPI",
        "description": "Crear estructura y dependencias",
        "status": "completed",
        "priority": "high",
        "created_at": _parse_dt("2025-03-01T10:00:00"),
    },
    {
        "id": 2,
        "title": "Implementar CRUD de tareas",
        "description": "GET, POST, PUT, PATCH, DELETE",
        "status": "in_progress",
        "priority": "high",
        "created_at": _parse_dt("2025-03-02T09:30:00"),
    },
    {
        "id": 3,
        "title": "Agregar filtros y paginación",
        "description": "",
        "status": "pending",
        "priority": "medium",
        "created_at": _parse_dt("2025-03-03T14:00:00"),
    },
]


def next_id() -> int:
    return max((t["id"] for t in tasks_db), default=0) + 1


def find_task(task_id: int) -> dict | None:
    return next((t for t in tasks_db if t["id"] == task_id), None)


# --- App FastAPI ---

app = FastAPI(
    title="To-Do List API",
    description="API CRUD de tareas. Módulo 6 — FastAPI Fundamentals.",
    version="1.0.0",
)


@app.get("/")
def root():
    return {
        "service": "To-Do List API",
        "version": "1.0.0",
        "total_tasks": len(tasks_db),
    }


@app.get("/tasks", response_model=TaskListResponse)
def list_tasks(
    status: Optional[str] = Query(None, description="Filtrar por estado"),
    priority: Optional[str] = Query(None, description="Filtrar por prioridad"),
    search: Optional[str] = Query(None, description="Buscar en título"),
    sort_by: Literal["created_at", "priority", "title"] = Query(
        default="created_at", description="Campo para ordenar"
    ),
    order: Literal["asc", "desc"] = Query(default="desc", description="Orden ascendente o descendente"),
    skip: int = Query(0, ge=0, description="Offset para paginación"),
    limit: int = Query(10, ge=1, le=100, description="Máximo items por página"),
):
    result = list(tasks_db)

    if status is not None:
        result = [t for t in result if t.get("status") == status]
    if priority is not None:
        result = [t for t in result if t.get("priority") == priority]
    if search is not None and search.strip():
        search_lower = search.lower().strip()
        result = [t for t in result if search_lower in t.get("title", "").lower()]

    total = len(result)

    reverse = order.lower() == "desc"
    if sort_by == "created_at":
        result.sort(key=lambda t: t.get("created_at", ""), reverse=reverse)
    elif sort_by == "priority":
        result.sort(
            key=lambda t: PRIORITY_ORDER.get(t.get("priority", ""), 3),
            reverse=reverse,
        )
    elif sort_by == "title":
        result.sort(key=lambda t: t.get("title", "").lower(), reverse=reverse)

    paginated = result[skip : skip + limit]

    return TaskListResponse(total=total, skip=skip, limit=limit, items=paginated)


@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    task = find_task(task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return task


@app.post("/tasks", status_code=201, response_model=TaskResponse)
def create_task(task: TaskCreate):
    data = task.model_dump()
    data["id"] = next_id()
    data["created_at"] = datetime.utcnow()
    tasks_db.append(data)
    return data


@app.put("/tasks/{task_id}", response_model=TaskResponse)
def update_task(task_id: int, task: TaskCreate):
    existing = find_task(task_id)
    if existing is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    updated = task.model_dump()
    updated["id"] = task_id
    updated["created_at"] = existing["created_at"]
    idx = tasks_db.index(existing)
    tasks_db[idx] = updated
    return updated


@app.patch("/tasks/{task_id}", response_model=TaskResponse)
def partial_update_task(task_id: int, updates: TaskUpdate):
    existing = find_task(task_id)
    if existing is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    update_data = updates.model_dump(exclude_unset=True)
    existing.update(update_data)
    return existing


@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    task = find_task(task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    tasks_db.remove(task)
    return {"message": "Task deleted", "id": task_id}

---

Decisiones de diseño

Por qué TaskListResponse envuelve items con total, skip y limit

El cliente necesita saber:

• cuántos items hay en total (total) para mostrar "mostrando X de Y" o el número de páginas
• en qué offset está (skip) para construir "siguiente" (skip + limit)
• cuántos items recibe en esta petición (limit)

Si solo devolvieras items, no podría saber si hay más páginas ni cuántas. Con total puede calcular (total + limit - 1) // limit para el número de páginas.

Por qué sort_by por defecto es created_at

Las tareas suelen consultarse por "últimas creadas" o "más recientes". Ordenar por created_at desc por defecto encaja con ese caso de uso. El usuario puede cambiar el orden explícitamente si quiere.

Por qué limit por defecto es 10

10 es un compromiso razonable entre cantidad y rendimiento. 100 o "sin límite" pueden ser peligrosos en datos grandes. Si necesitas más, el cliente puede enviar limit=50 o limit=100 (con le=100 lo acotas en la validación).

Por qué filtro + orden + paginación en un solo endpoint

Separar en /tasks/filtered, /tasks/search y /tasks/paginated complicaría la API y el cliente tendría que encadenar varias llamadas. Un único endpoint con query params permite combinar todo: ?status=pending&search=api&sort_by=priority&skip=0&limit=10. Es el enfoque estándar en REST.

---

Edge cases

¿Qué pasa si skip > total?

Slicing list[skip:skip+limit] devuelve [] sin error. items queda vacío, total sigue siendo el total correcto (antes de paginar). El cliente puede interpretar "no hay más resultados".

¿Qué pasa si limit=0?

Con Query(10, ge=1, le=100) limit no puede ser 0: FastAPI devuelve 422. Si no usaras ge=1, result[skip:skip+0] daría [], que es coherente pero poco útil; es preferible rechazar limit=0.

¿Qué pasa si sort_by es un campo inválido?

Con Literal["created_at", "priority", "title"] FastAPI rechaza cualquier otro valor con 422. Si usaras str sin validar, un sort_by=invalid llevaría al elif de title o a un fallo; con Literal evitas ese caso.

¿Qué pasa si search es vacío o solo espacios?

La condición if search is not None and search.strip() evita aplicar búsqueda cuando search="" o search=" ". En ese caso se devuelven todas las tareas (tras filtros). Si no comprobases .strip(), una búsqueda de espacios podría no encontrar nada de forma inesperada.

---

Errores comunes

1. Olvidar hacer copia de la lista antes de filtrar — Si filtras sobre tasks_db directamente (tasks_db = [t for t in tasks_db if ...]), mutas la base de datos. Usa result = list(tasks_db) y trabaja sobre result.

2. Calcular total después de paginar — total debe ser el número de resultados tras filtros y búsqueda, antes de paginar. Si calculas total = len(paginated), pierdes la info del total real.

3. No tratar search vacío — Si usas if search: sin comprobar search.strip(), un valor como " " se interpreta como truthy y aplicas una búsqueda que no devuelve nada. Siempre valida search.strip().

4. Ordenar priority como string — Ordenar por t["priority"] alfabéticamente da "high" después de "low". Usa siempre PRIORITY_ORDER o un mapeo explícito.

5. Usar page sin convertir a skip — Si recibes page del cliente, conviértelo a offset: skip = (page - 1) * limit. No apliques page directamente como índice.

---

Troubleshooting

1. Los filtros no devuelven nada cuando deberían — Revisa que los valores coincidan exactamente (status: "in_progress" con guion bajo, no "in progress"). Verifica en /docs qué valor llega al endpoint.

2. La búsqueda no encuentra tareas que sí contienen el texto — Comprueba que usas .lower() en ambos lados y que buscas en el campo correcto. Si la búsqueda está en description pero tu código solo mira title, no coincidirá.

3. Paginación devuelve duplicados o salta items — Si los datos cambian entre peticiones (crear/borrar tareas), el offset puede desplazarse. Para datos muy dinámicos, considera cursor-based pagination en lugar de skip/limit.

4. 422 con sort_by u order — Si usas Literal, el cliente debe enviar exactamente uno de los valores permitidos. Comprueba en la documentación qué opciones son válidas.

---

Cuando migres a SQL

La misma lógica se traslada directamente a bases de datos. En PostgreSQL, por ejemplo:

SELECT * FROM tasks
WHERE ($1::text IS NULL OR status = $1)
  AND ($2::text IS NULL OR priority = $2)
  AND ($3::text IS NULL OR LOWER(title) LIKE '%' || LOWER($3) || '%')
ORDER BY
  CASE WHEN $4 = 'created_at' THEN created_at END DESC NULLS LAST,
  CASE WHEN $4 = 'priority' THEN priority_order END ASC,
  CASE WHEN $4 = 'title' THEN LOWER(title) END ASC
LIMIT $5 OFFSET $6;

El orden sigue siendo: filtrar → ordenar → paginar. Con SQLAlchemy u otro ORM expresarás esto con .filter(), .order_by(), .limit() y .offset(). El patrón conceptual no cambia.

---

Ejemplos de consultas

GET /tasks
GET /tasks?status=pending
GET /tasks?priority=high
GET /tasks?search=crud
GET /tasks?sort_by=priority&order=asc
GET /tasks?skip=0&limit=2
GET /tasks?status=in_progress&sort_by=created_at&order=desc&limit=5

---

Validación de skip y limit

Usando Query(0, ge=0) y Query(10, ge=1, le=100) evitas skip negativo o limit fuera de rango. El cliente recibe 422 si envía valores inválidos.

---

Checklist

• ✅ Filtro por status
• ✅ Filtro por priority
• ✅ Búsqueda por texto en title (case insensitive)
• ✅ Ordenamiento por created_at, priority, title
• ✅ Orden asc/desc
• ✅ Paginación skip/limit
• ✅ Respuesta con total, skip, limit, items

---

Consejo para el frontend

Cuando consumas este endpoint desde React, Vue o cualquier SPA, guarda en estado total, skip y limit además de items. Con eso puedes:

• Mostrar "Página X de Y" usando (total + limit - 1) // limit
• Habilitar o deshabilitar el botón "Siguiente" si skip + limit >= total
• Construir la URL de la siguiente página: ?skip=${skip + limit}&limit=${limit}&status=${status}&... (repite los mismos filtros para mantener el contexto)

---

Resumen

En esta cápsula has implementado el patrón típico de listados en REST:

• Filtros opcionales (status, priority) con Optional[str] = None para no aplicarlos si no se envían
• Búsqueda por texto en title, case insensitive, ignorando cadenas vacías o solo espacios
• Ordenamiento con PRIORITY_ORDER para prioridad y sort_by/order para el resto de campos
• Paginación con skip/limit y respuesta con total, skip, limit e items
• Validación con Query y Literal para evitar valores inválidos
• Edge cases contemplados: skip > total, search vacío, limit acotado

Con esto, GET /tasks se comporta como los listados de APIs profesionales y el cliente puede construir UIs paginadas, filtradas y ordenadas con una sola ruta.

---

Recursos adicionales

• FastAPI — Query Parameters — Documentación oficial de query params
• REST API Design: Filtering, Sorting, Pagination — Patrones de diseño para listados
• Stripe API — Pagination — Cómo Stripe maneja limit y starting_after
• GitHub REST API — Pagination — Paginación con per_page y page
• PostgreSQL LIMIT and OFFSET — Equivalente en SQL para cuando uses base de datos
• Cursor vs Offset Pagination — Comparación cuando migres a SQL

---

Próximo paso

En la Cápsula 05 agregarás error handling profesional (handlers personalizados, formato consistente), CORS middleware, metadata mejorada para la documentación y el pulido final del API.

---

Módulo 6, Cápsula 04 — FastAPI Fundamentals Guide