Module 6: Project — CRUD API (To-Do List)

Paso 2: Endpoints CRUD

Next →

Descripción

Los cimientos están listos: modelos Pydantic definidos, datos iniciales cargados, app configurada. Ahora viene la parte central del proyecto: implementar todos los endpoints CRUD que hacen funcionar la API.

Al terminar esta cápsula tendrás:

  • POST /tasks — crear tareas con ID auto-generado y timestamps
  • GET /tasks — listar tareas con filtros (status, priority, search) y paginación (skip/limit)
  • GET /tasks/{task_id} — obtener una tarea por ID
  • PUT /tasks/{task_id} — actualización completa
  • PATCH /tasks/{task_id} — actualización parcial
  • DELETE /tasks/{task_id} — eliminar una tarea

Todos los endpoints usan los modelos Pydantic que definiste en la cápsula anterior, lanzan HTTPException cuando algo sale mal, y retornan status codes HTTP correctos.

El archivo completo: app/routes.py

Este es el código completo del router con todos los endpoints. Después del código, cada endpoint se explica en detalle.

# app/routes.py
from fastapi import APIRouter, HTTPException, Query, Path
from starlette import status

from app.models import (
    Status,
    Priority,
    TaskCreate,
    TaskUpdate,
    TaskPatch,
    TaskResponse,
    TaskListResponse,
)
from app.data import tasks, generate_id, get_current_timestamp, find_task


router = APIRouter()


# --- General ---

@router.get("/health", tags=["General"], summary="Health check")
def health_check():
    """Verificar que la API está funcionando."""
    return {
        "status": "healthy",
        "service": "To-Do List API",
        "version": "1.0.0",
    }


# --- Create ---

@router.post(
    "/tasks",
    response_model=TaskResponse,
    status_code=status.HTTP_201_CREATED,
    tags=["Tasks"],
    summary="Create a new task",
)
def create_task(task: TaskCreate):
    """Crear una tarea nueva con ID y timestamps auto-generados."""
    now = get_current_timestamp()

    new_task = {
        "id": generate_id(),
        **task.model_dump(),
        "created_at": now,
        "updated_at": now,
    }

    tasks.append(new_task)
    return new_task


# --- Read (list) ---

@router.get(
    "/tasks",
    response_model=TaskListResponse,
    tags=["Tasks"],
    summary="List tasks with filters and pagination",
)
def list_tasks(
    task_status: Status | None = Query(
        default=None,
        alias="status",
        description="Filtrar por estado",
    ),
    priority: Priority | None = Query(
        default=None,
        description="Filtrar por prioridad",
    ),
    search: str | None = Query(
        default=None,
        min_length=1,
        max_length=100,
        description="Buscar en título y descripción",
    ),
    skip: int = Query(
        default=0,
        ge=0,
        description="Resultados a saltar (paginación)",
    ),
    limit: int = Query(
        default=10,
        ge=1,
        le=100,
        description="Máximo de resultados por página (1-100)",
    ),
):
    """Listar tareas con filtros opcionales y paginación."""
    results = tasks.copy()

    if task_status is not None:
        results = [t for t in results if t["status"] == task_status.value]

    if priority is not None:
        results = [t for t in results if t["priority"] == priority.value]

    if search is not None:
        query = search.lower()
        results = [
            t for t in results
            if query in t["title"].lower()
            or query in t["description"].lower()
        ]

    total = len(results)
    paginated = results[skip : skip + limit]

    return TaskListResponse(
        total=total,
        count=len(paginated),
        skip=skip,
        limit=limit,
        data=[TaskResponse(**t) for t in paginated],
    )


# --- Read (single) ---

@router.get(
    "/tasks/{task_id}",
    response_model=TaskResponse,
    tags=["Tasks"],
    summary="Get task by ID",
)
def get_task(
    task_id: int = Path(..., ge=1, description="ID de la tarea"),
):
    """Obtener una tarea por su ID."""
    task = find_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found",
        )
    return task


# --- Update (full) ---

@router.put(
    "/tasks/{task_id}",
    response_model=TaskResponse,
    tags=["Tasks"],
    summary="Full update of a task",
)
def update_task(
    task_data: TaskUpdate,
    task_id: int = Path(..., ge=1, description="ID de la tarea a actualizar"),
):
    """Reemplazar todos los campos de una tarea (PUT)."""
    task = find_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found",
        )

    index = tasks.index(task)
    tasks[index] = {
        "id": task_id,
        **task_data.model_dump(),
        "created_at": task["created_at"],
        "updated_at": get_current_timestamp(),
    }

    return tasks[index]


# --- Update (partial) ---

@router.patch(
    "/tasks/{task_id}",
    response_model=TaskResponse,
    tags=["Tasks"],
    summary="Partial update of a task",
)
def patch_task(
    task_data: TaskPatch,
    task_id: int = Path(
        ..., ge=1, description="ID de la tarea a actualizar parcialmente"
    ),
):
    """Actualizar solo los campos enviados (PATCH)."""
    task = find_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found",
        )

    update_data = task_data.model_dump(exclude_unset=True)
    if not update_data:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="No fields provided for update",
        )

    task.update(update_data)
    task["updated_at"] = get_current_timestamp()

    return task


# --- Delete ---

@router.delete(
    "/tasks/{task_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    tags=["Tasks"],
    summary="Delete a task",
)
def delete_task(
    task_id: int = Path(..., ge=1, description="ID de la tarea a eliminar"),
):
    """Eliminar una tarea por su ID."""
    task = find_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found",
        )

    tasks.remove(task)

Explicación endpoint por endpoint

POST /tasks — Crear tarea

@router.post(
    "/tasks",
    response_model=TaskResponse,
    status_code=status.HTTP_201_CREATED,
    tags=["Tasks"],
    summary="Create a new task",
)
def create_task(task: TaskCreate):
    now = get_current_timestamp()

    new_task = {
        "id": generate_id(),
        **task.model_dump(),
        "created_at": now,
        "updated_at": now,
    }

    tasks.append(new_task)
    return new_task

¿Qué hace?

  1. Recibe un body que Pydantic valida contra TaskCreate
  2. Genera un ID único con generate_id()
  3. Crea un timestamp para created_at y updated_at (ambos iguales al crear)
  4. task.model_dump() convierte el modelo Pydantic a dict: {"title": "...", "description": "...", "priority": "medium", "status": "pending"}
  5. **task.model_dump() desempaqueta esos campos en el dict del nuevo task
  6. Retorna la tarea creada con status 201 (Created)

¿Por qué status_code=201?

HTTP 201 significa "se creó un recurso nuevo." Es la convención para POST exitoso. Sin especificarlo, FastAPI usaría 200, que es correcto pero menos semántico.

¿Por qué response_model=TaskResponse?

Garantiza que la respuesta incluye id, created_at y updated_at — campos que el cliente no envió pero que la API generó. También actúa como documentación: en /docs el cliente sabe exactamente qué estructura esperar.

¿Por qué model_dump() y no .dict()?

.dict() es Pydantic v1. model_dump() es Pydantic v2. Hacen lo mismo (convertir modelo a dict), pero model_dump() es la forma actual. Si usas .dict(), funciona pero Pydantic muestra un deprecation warning.

GET /tasks — Listar con filtros y paginación

@router.get(
    "/tasks",
    response_model=TaskListResponse,
    tags=["Tasks"],
    summary="List tasks with filters and pagination",
)
def list_tasks(
    task_status: Status | None = Query(
        default=None,
        alias="status",
        description="Filtrar por estado",
    ),
    priority: Priority | None = Query(
        default=None,
        description="Filtrar por prioridad",
    ),
    search: str | None = Query(
        default=None,
        min_length=1,
        max_length=100,
        description="Buscar en título y descripción",
    ),
    skip: int = Query(default=0, ge=0, description="Resultados a saltar"),
    limit: int = Query(default=10, ge=1, le=100, description="Máximo de resultados"),
):
    results = tasks.copy()

    if task_status is not None:
        results = [t for t in results if t["status"] == task_status.value]

    if priority is not None:
        results = [t for t in results if t["priority"] == priority.value]

    if search is not None:
        query = search.lower()
        results = [
            t for t in results
            if query in t["title"].lower()
            or query in t["description"].lower()
        ]

    total = len(results)
    paginated = results[skip : skip + limit]

    return TaskListResponse(
        total=total,
        count=len(paginated),
        skip=skip,
        limit=limit,
        data=[TaskResponse(**t) for t in paginated],
    )

¿Por qué task_status con alias="status"?

status es una palabra reservada en Python (no técnicamente, pero FastAPI ya importa status de starlette para los códigos HTTP). Además, el parámetro en Python no puede llamarse status si ya tienes un import con ese nombre. El alias="status" hace que en la URL se use ?status=pending pero en el código Python se llame task_status.

¿Qué es .value en task_status.value?

Cuando el enum Status llega como query param, FastAPI lo convierte a la instancia del enum: Status.pending. Pero los datos en la lista son strings: "pending". .value extrae el string del enum para comparar: Status.pending.value == "pending".

¿Cómo funcionan los filtros?

Cada filtro es independiente y se aplica en secuencia:

Todos los tasks (5)
    → Filtrar por status (si se proporcionó)
    → Filtrar por priority (si se proporcionó)
    → Filtrar por search (si se proporcionó)
    → total = len(resultados filtrados)
    → Paginar con skip/limit
    → count = len(resultados paginados)

Los filtros son aditivos: si filtras por status=pending Y priority=high, obtienes tareas que son pending Y high. No pending O high.

¿Cómo funciona la paginación?

skip y limit actúan sobre el resultado filtrado:

paginated = results[skip : skip + limit]
# skip=0, limit=10 → results[0:10]  → primeros 10
# skip=10, limit=10 → results[10:20] → siguientes 10
# skip=20, limit=5  → results[20:25] → 5 después del 20

total es la cantidad total de resultados filtrados (antes de paginar). count es la cantidad de resultados en esta página. El cliente compara total con skip + count para saber si hay más páginas.

¿Cómo funciona la búsqueda?

search busca en título y descripción simultáneamente, case-insensitive:

query = search.lower()
results = [
    t for t in results
    if query in t["title"].lower()
    or query in t["description"].lower()
]

GET /tasks?search=api encontraría tareas con "API" en el título o en la descripción.

GET /tasks/{task_id} — Obtener tarea por ID

@router.get(
    "/tasks/{task_id}",
    response_model=TaskResponse,
    tags=["Tasks"],
    summary="Get task by ID",
)
def get_task(
    task_id: int = Path(..., ge=1, description="ID de la tarea"),
):
    task = find_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found",
        )
    return task

El patrón: buscar → verificar → responder.

  1. find_task(task_id) busca en la lista. Retorna el dict o None
  2. Si es None, lanza HTTPException con 404
  3. Si existe, retorna el task (FastAPI lo serializa con TaskResponse)

¿Por qué Path(..., ge=1)?

  • ... significa que el parámetro es requerido (en path params siempre lo es, pero es explícito)
  • ge=1 valida que el ID sea >= 1. Si alguien pide GET /tasks/0 o GET /tasks/-5, FastAPI retorna 422 automáticamente

PUT /tasks/{task_id} — Actualización completa

@router.put(
    "/tasks/{task_id}",
    response_model=TaskResponse,
    tags=["Tasks"],
    summary="Full update of a task",
)
def update_task(
    task_data: TaskUpdate,
    task_id: int = Path(..., ge=1, description="ID de la tarea a actualizar"),
):
    task = find_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found",
        )

    index = tasks.index(task)
    tasks[index] = {
        "id": task_id,
        **task_data.model_dump(),
        "created_at": task["created_at"],
        "updated_at": get_current_timestamp(),
    }

    return tasks[index]

PUT reemplaza el recurso completo. El cliente envía todos los campos y la API reemplaza la tarea entera. Solo se preservan:

  • id — nunca cambia
  • created_at — la fecha de creación es inmutable
  • updated_at — se actualiza al momento del cambio

¿Por qué tasks.index(task) y no modificar task directamente?

Porque con PUT reemplazamos el dict entero en la lista, no modificamos campos individuales. tasks.index(task) encuentra la posición del dict original, y luego asignamos un dict completamente nuevo en esa posición.

PATCH /tasks/{task_id} — Actualización parcial

@router.patch(
    "/tasks/{task_id}",
    response_model=TaskResponse,
    tags=["Tasks"],
    summary="Partial update of a task",
)
def patch_task(
    task_data: TaskPatch,
    task_id: int = Path(
        ..., ge=1, description="ID de la tarea a actualizar parcialmente"
    ),
):
    task = find_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found",
        )

    update_data = task_data.model_dump(exclude_unset=True)
    if not update_data:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="No fields provided for update",
        )

    task.update(update_data)
    task["updated_at"] = get_current_timestamp()

    return task

PATCH actualiza solo los campos enviados. Si el cliente envía {"status": "completed"}, solo cambia status — el resto permanece intacto.

La clave: model_dump(exclude_unset=True)

# Cliente envía: {"status": "completed"}
task_data.model_dump()
# → {"title": None, "description": None, "priority": None, "status": "completed"}
# ❌ Sobreescribiría title, description, priority con None

task_data.model_dump(exclude_unset=True)
# → {"status": "completed"}
# ✅ Solo incluye los campos que el cliente envió

Sin exclude_unset=True, PATCH sobreescribiría todos los campos no enviados con None. Con exclude_unset=True, solo incluye los campos que el cliente incluyó en el body.

¿Por qué 400 si el body está vacío?

Un PATCH sin campos no tiene sentido — el cliente pidió actualizar pero no dijo qué. Es un error del cliente (400), no un "no encontrado" (404) ni un conflicto (409).

¿Por qué task.update(update_data) y no crear dict nuevo?

Porque en PATCH queremos modificar el dict existente, no reemplazarlo. task.update({"status": "completed"}) solo cambia status y deja todos los demás campos intactos. Luego actualizamos updated_at manualmente.

DELETE /tasks/{task_id} — Eliminar tarea

@router.delete(
    "/tasks/{task_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    tags=["Tasks"],
    summary="Delete a task",
)
def delete_task(
    task_id: int = Path(..., ge=1, description="ID de la tarea a eliminar"),
):
    task = find_task(task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found",
        )

    tasks.remove(task)

¿Por qué 204 y no 200?

HTTP 204 (No Content) significa "la operación fue exitosa, no hay datos que retornar." Es la convención para DELETE: el recurso ya no existe, no tiene sentido retornarlo. La función no tiene return — FastAPI envía una respuesta vacía con status 204.

¿Por qué la función no retorna nada?

Con status_code=204, FastAPI ignora el valor de retorno y envía una respuesta sin body. Si intentas retornar algo con 204, FastAPI podría lanzar un warning o error dependiendo de la configuración.

Verificación: probar todos los endpoints

1. Crear una tarea (POST)

curl -s -w "\nHTTP: %{http_code}\n" -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Estudiar Python avanzado", "description": "Decoradores, generators, async", "priority": "high"}'

Respuesta esperada (HTTP 201):

{
    "id": 6,
    "title": "Estudiar Python avanzado",
    "description": "Decoradores, generators, async",
    "status": "pending",
    "priority": "high",
    "created_at": "2026-03-13T...",
    "updated_at": "2026-03-13T..."
}
  • id es 6 (las 5 precargadas ocupan 1-5)
  • status defaulteó a "pending" porque no se envió
  • created_at y updated_at son iguales (recién creado)

2. Crear tarea mínima (solo título)

curl -s -w "\nHTTP: %{http_code}\n" -X POST http://127.0.0.1:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Tarea rápida"}'
{
    "id": 7,
    "title": "Tarea rápida",
    "description": "",
    "status": "pending",
    "priority": "medium",
    "created_at": "2026-03-13T...",
    "updated_at": "2026-03-13T..."
}

Solo title es requerido. description defaultea a "", priority a "medium", status a "pending".

3. Listar todas las tareas (GET)

curl -s http://127.0.0.1:8000/tasks | python -m json.tool

Retorna "total": 7 (5 precargadas + 2 que creaste), "count": 7 (todas caben en una página), "data": [...] con las 7 tareas.

4. Filtrar por status

curl -s "http://127.0.0.1:8000/tasks?status=pending" | python -m json.tool

Solo retorna tareas con status "pending".

5. Filtrar por priority

curl -s "http://127.0.0.1:8000/tasks?priority=high" | python -m json.tool

Solo retorna tareas con priority "high".

6. Combinar filtros

curl -s "http://127.0.0.1:8000/tasks?status=pending&priority=high" | python -m json.tool

Solo tareas que son pending Y high.

7. Buscar texto

curl -s "http://127.0.0.1:8000/tasks?search=api" | python -m json.tool

Retorna tareas que contienen "api" en el título o descripción (case-insensitive).

8. Paginación

# Primera página (3 resultados)
curl -s "http://127.0.0.1:8000/tasks?limit=3" | python -m json.tool

# Segunda página
curl -s "http://127.0.0.1:8000/tasks?skip=3&limit=3" | python -m json.tool

9. Obtener tarea por ID (GET)

curl -s -w "\nHTTP: %{http_code}\n" http://127.0.0.1:8000/tasks/1

Retorna la tarea 1 con HTTP 200.

10. Tarea inexistente (404)

curl -s -w "\nHTTP: %{http_code}\n" http://127.0.0.1:8000/tasks/999
{"detail": "Task with id 999 not found"}

HTTP 404.

11. PUT, PATCH, DELETE

Prueba las actualizaciones y eliminación con los mismos patrones. Verifica que:

  • PUT reemplaza todos los campos y actualiza updated_at
  • PATCH con {"status": "completed"} solo cambia ese campo
  • PATCH con {} retorna 400 ("No fields provided")
  • DELETE retorna 204 sin body
  • GET después de DELETE retorna 404

Troubleshooting

Problema 1: ImportError: cannot import name 'Status' from 'app.models'

Causa: models.py no define Status o tiene un error de sintaxis que impide el import.

Verifica que app/models.py contiene class Status(str, Enum): y que no hay errores de indentación.

Problema 2: Los filtros no funcionan — siempre retorna todas las tareas

Causa: La comparación entre el enum y el string falla.

# ❌ Esto no coincide: Status.pending != "pending"
results = [t for t in results if t["status"] == task_status]

# ✅ Esto sí: "pending" == "pending"
results = [t for t in results if t["status"] == task_status.value]

Usa .value para extraer el string del enum.

Problema 3: PATCH sobreescribe campos con None

Causa: Falta exclude_unset=True.

# ❌ Incluye todos los campos (los no enviados son None)
update_data = task_data.model_dump()

# ✅ Solo incluye los campos que el cliente envió
update_data = task_data.model_dump(exclude_unset=True)

Problema 4: DELETE retorna body cuando debería ser 204 sin body

Causa: La función tiene un return.

# ❌ Retorna body con 204 — puede causar warnings
tasks.remove(task)
return {"message": "deleted"}

# ✅ Sin return — 204 no tiene body
tasks.remove(task)

Problema 5: GET /tasks/stats retorna 422 en vez de stats

Causa: FastAPI interpreta stats como el valor de {task_id}. El orden de las rutas importa: /tasks/stats debe estar antes de /tasks/{task_id}.

# ✅ stats primero
@router.get("/tasks/stats")
def task_stats(): ...

# Después el path param
@router.get("/tasks/{task_id}")
def get_task(task_id: int): ...

Resumen

  • Implementaste 6 endpoints CRUD en app/routes.py usando APIRouter
  • POST /tasks crea tareas con ID y timestamps auto-generados (201)
  • GET /tasks soporta filtros por status, priority, search y paginación con skip/limit
  • GET /tasks/{task_id} retorna la tarea o 404
  • PUT /tasks/{task_id} reemplaza completa la tarea, preservando id y created_at
  • PATCH /tasks/{task_id} usa model_dump(exclude_unset=True) para actualizar solo los campos enviados
  • DELETE /tasks/{task_id} elimina con status 204 (No Content)
  • El patrón "buscar → verificar → actuar" se repite en todos los endpoints y los filtros son aditivos (AND)

Recursos adicionales

  1. FastAPI - Path Parameters — Path params con validación
  2. FastAPI - Query Parameters — Query params opcionales y validación
  3. FastAPI - Request Body — Modelos Pydantic como body
  4. FastAPI - Response Model — Controlar la respuesta
  5. FastAPI - APIRouter — Organizar endpoints en archivos separados
  6. Pydantic v2 - model_dump — Serialización con opciones
← PreviousPaso 3: Filtros, Búsqueda y PaginaciónNext →Paso 4: Error Handling, CORS y Documentación