Modelos Pydantic y Capa de Datos

Descripción

En la cápsula anterior creaste la estructura de carpetas del proyecto: todo-api/app/ con __init__.py, main.py, models.py, routes.py y data.py. La estructura está lista pero los archivos están vacíos. Ahora toca llenarlos con lo que define la personalidad de tu API: los modelos de datos y el almacenamiento.

Los modelos Pydantic son los contratos de tu API. Definen exactamente qué datos acepta cada endpoint, qué valida, y qué retorna. Sin modelos bien definidos, tu API sería un caos de dicts sin estructura. Con ellos, FastAPI genera documentación automática, valida inputs antes de que tu código los toque, y serializa outputs de forma consistente.

Aquí vas a definir la entidad Task, crear 6 modelos Pydantic con herencia, implementar validaciones custom, y construir la capa de datos in-memory. La clave: intenta escribir el código tú primero antes de abrir los hints. Ya aprendiste Pydantic en el Módulo 4 — ahora lo aplicas a un proyecto real.

El dominio de datos: Task

Cada tarea en tu To-Do API tiene estos campos:

Decisiones clave: status y priority usan Literal en vez de Enum — más ligero para conjuntos fijos de strings. Timestamps como strings ISO 8601 simplifican la serialización JSON. updated_at empieza en None (nunca modificado) y se actualiza en PUT/PATCH.

Especificación de modelos — app/models.py

Tu archivo necesita 6 modelos con esta jerarquía:

TaskBase (campos compartidos + validación)
├── TaskCreate (hereda, title requerido, resto con defaults)
├── TaskUpdate (hereda, todos los campos requeridos)
└── TaskResponse (hereda, agrega id + timestamps)

TaskPatch (independiente, todo Optional)
TaskListResponse (wrapper de lista + count)

¿Por qué TaskBase? Porque title, description, status y priority aparecen en varios modelos con las mismas restricciones. Sin base, duplicarías validación en 3 lugares. Con TaskBase, un cambio se propaga a todos.

Modelo 1: TaskBase

Especificación: Hereda de BaseModel. Campos: title (str, 1-200 chars), description (str | None, max 1000, default None), status (solo "pending"/"in_progress"/"completed", default "pending"), priority (solo "low"/"medium"/"high"/"urgent", default "medium"). Cada campo con description y examples en Field().

Pistas: from typing import Literal para restringir status/priority. Define type aliases (ALLOWED_STATUSES, ALLOWED_PRIORITIES) para reutilizar en TaskPatch.

from typing import Literal

from pydantic import BaseModel, Field, field_validator


ALLOWED_STATUSES = Literal["pending", "in_progress", "completed"]
ALLOWED_PRIORITIES = Literal["low", "medium", "high", "urgent"]


class TaskBase(BaseModel):
    """Campos compartidos entre modelos de Task."""

    title: str = Field(
        min_length=1,
        max_length=200,
        description="Título de la tarea",
        examples=["Revisar pull request del módulo de auth"],
    )
    description: str | None = Field(
        default=None,
        max_length=1000,
        description="Descripción detallada de la tarea",
        examples=["Revisar cambios en el sistema de login, verificar tokens JWT"],
    )
    status: ALLOWED_STATUSES = Field(
        default="pending",
        description="Estado actual de la tarea",
        examples=["pending"],
    )
    priority: ALLOWED_PRIORITIES = Field(
        default="medium",
        description="Nivel de prioridad",
        examples=["high"],
    )

Modelo 2: TaskCreate

Especificación: Hereda de TaskBase. El cliente envía estos datos para crear una tarea. title requerido (de TaskBase), resto con defaults. No incluye id, created_at ni updated_at — los genera el servidor. Si hereda todo y no necesita nada extra... ¿qué va en el body?

class TaskCreate(TaskBase):
    """Modelo para crear una tarea nueva.

    Hereda todos los campos de TaskBase.
    id, created_at y updated_at se generan en el servidor.
    """

    pass

Modelo 3: TaskUpdate

Especificación: Hereda de TaskBase. Para PUT: todos los campos son requeridos. title ya es requerido ✅. Pero description, status y priority tienen defaults — necesitas eliminarlos.

Desafío: ¿Cómo haces que campos con default en la clase padre sean requeridos en la hija? Pista: Field(...) (Ellipsis = requerido).

class TaskUpdate(TaskBase):
    """Modelo para reemplazo completo (PUT).

    Todos los campos son requeridos porque PUT
    reemplaza el recurso entero.
    """

    description: str | None = Field(
        ...,
        max_length=1000,
        description="Descripción detallada de la tarea",
    )
    status: ALLOWED_STATUSES = Field(
        ...,
        description="Estado actual de la tarea",
    )
    priority: ALLOWED_PRIORITIES = Field(
        ...,
        description="Nivel de prioridad",
    )

Modelo 4: TaskPatch

Especificación: No hereda de TaskBase (independiente). Para PATCH: todos los campos Optional (default None). Campos: title, description, status, priority con mismas restricciones. El endpoint usará model_dump(exclude_unset=True). No hereda porque title es requerido en TaskBase — redefinir todo anula el beneficio.

class TaskPatch(BaseModel):
    """Modelo para actualización parcial (PATCH).

    Todos los campos son opcionales. Solo los campos
    incluidos en el request se actualizan.
    """

    title: str | None = Field(
        default=None,
        min_length=1,
        max_length=200,
        description="Nuevo título",
    )
    description: str | None = Field(
        default=None,
        max_length=1000,
        description="Nueva descripción",
    )
    status: ALLOWED_STATUSES | None = Field(
        default=None,
        description="Nuevo estado",
    )
    priority: ALLOWED_PRIORITIES | None = Field(
        default=None,
        description="Nueva prioridad",
    )

Modelo 5: TaskResponse

Especificación: Hereda de TaskBase. Agrega: id (int), created_at (str), updated_at (str | None). Incluye model_config con json_schema_extra para ejemplo en /docs. Se usa como response_model en todos los endpoints.

class TaskResponse(TaskBase):
    """Modelo de respuesta — incluye campos auto-generados."""

    id: int
    created_at: str = Field(
        description="Fecha de creación (ISO 8601)",
        examples=["2026-03-13T14:30:00+00:00"],
    )
    updated_at: str | None = Field(
        default=None,
        description="Fecha de última modificación (ISO 8601)",
        examples=["2026-03-13T15:45:00+00:00"],
    )

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "id": 1,
                    "title": "Revisar pull request",
                    "description": "Revisar PR #42 del módulo de auth",
                    "status": "in_progress",
                    "priority": "high",
                    "created_at": "2026-03-13T10:00:00+00:00",
                    "updated_at": "2026-03-13T14:30:00+00:00",
                }
            ]
        }
    }

Modelo 6: TaskListResponse

Especificación: Independiente (BaseModel). Campos: tasks (list[TaskResponse]), count (int — cantidad de tareas).

class TaskListResponse(BaseModel):
    """Respuesta de lista de tareas con metadata."""

    tasks: list[TaskResponse] = Field(
        description="Lista de tareas",
    )
    count: int = Field(
        description="Cantidad de tareas en la respuesta",
    )

Validators

Los constraints de Field() cubren longitudes y tipos, pero hay validaciones que necesitan lógica custom.

Especificación

Validator de title (en TaskBase):

• Aplica strip() al valor
• Si queda vacío después del strip, lanza ValueError("Title cannot be empty or whitespace only")
• Retorna el valor limpio
• Usa @field_validator("title") con @classmethod

Validator de description (en TaskBase):

• Si es None, retórnalo sin cambios
• Si es string, aplica strip() — si queda vacío, retorna None
• Retorna el valor limpio

TaskPatch necesita validators similares pero que manejen None (todos sus campos son opcionales).

@field_validator("title")
@classmethod
def title_must_not_be_whitespace(cls, v: str) -> str:
    if not v.strip():
        raise ValueError("Title cannot be empty or whitespace only")
    return v.strip()

@field_validator("description")
@classmethod
def clean_description(cls, v: str | None) -> str | None:
    if v is None:
        return v
    stripped = v.strip()
    return None if not stripped else stripped

@field_validator("title")
@classmethod
def title_must_not_be_whitespace(cls, v: str | None) -> str | None:
    if v is not None and not v.strip():
        raise ValueError("Title cannot be empty or whitespace only")
    return v.strip() if v else v

@field_validator("description")
@classmethod
def clean_description(cls, v: str | None) -> str | None:
    if v is None:
        return v
    stripped = v.strip()
    return None if not stripped else stripped

Almacenamiento en memoria — app/data.py

Tu API necesita dónde guardar las tareas. Para este proyecto usas una lista de dicts en memoria — cada restart vuelve al estado inicial, y eso está bien para aprender.

Especificación

1. Lista de tareas iniciales (tasks: list[dict]) con 6 tareas precargadas:

Cada dict debe tener todos los campos: id, title, description, status, priority, created_at, updated_at. Timestamps como strings ISO. Las tareas completadas y en progreso deben tener updated_at; las pendientes pueden tener None.

2. find_task(task_id: int) -> dict | None — Busca por ID, retorna dict o None. Pista: next() con generator expression.

3. generate_id() -> int — Retorna siguiente ID (contador global que incrementa). Inicia en 7.

4. get_current_timestamp() -> str — Retorna datetime.now(timezone.utc).isoformat().

# app/data.py
from datetime import datetime, timezone


tasks: list[dict] = [
    {
        "id": 1,
        "title": "Revisar pull request",
        "description": "Revisar PR #42: refactor del sistema de login con JWT",
        "status": "completed",
        "priority": "high",
        "created_at": "2026-03-10T09:00:00+00:00",
        "updated_at": "2026-03-11T16:45:00+00:00",
    },
    {
        "id": 2,
        "title": "Escribir documentación del API",
        "description": "Documentar endpoints, schemas y ejemplos de uso en README",
        "status": "in_progress",
        "priority": "medium",
        "created_at": "2026-03-11T10:30:00+00:00",
        "updated_at": "2026-03-12T11:00:00+00:00",
    },
    {
        "id": 3,
        "title": "Configurar CI/CD con GitHub Actions",
        "description": "Pipeline con lint, tests y deploy automático a staging",
        "status": "pending",
        "priority": "urgent",
        "created_at": "2026-03-12T08:00:00+00:00",
        "updated_at": None,
    },
    {
        "id": 4,
        "title": "Refactorizar módulo de autenticación",
        "description": "Separar lógica de tokens en su propio service, agregar refresh tokens",
        "status": "in_progress",
        "priority": "high",
        "created_at": "2026-03-12T09:15:00+00:00",
        "updated_at": "2026-03-13T10:20:00+00:00",
    },
    {
        "id": 5,
        "title": "Actualizar dependencias del proyecto",
        "description": None,
        "status": "pending",
        "priority": "low",
        "created_at": "2026-03-13T07:00:00+00:00",
        "updated_at": None,
    },
    {
        "id": 6,
        "title": "Escribir tests unitarios para models",
        "description": "Cobertura mínima del 80% en Pydantic models y validators",
        "status": "pending",
        "priority": "medium",
        "created_at": "2026-03-13T08:30:00+00:00",
        "updated_at": None,
    },
]


id_counter: int = 7


def generate_id() -> int:
    """Genera el siguiente ID disponible e incrementa el contador."""
    global id_counter
    current = id_counter
    id_counter += 1
    return current


def get_current_timestamp() -> str:
    """Retorna la fecha/hora actual en formato ISO 8601 UTC."""
    return datetime.now(timezone.utc).isoformat()


def find_task(task_id: int) -> dict | None:
    """Busca una tarea por ID. Retorna None si no existe."""
    return next((t for t in tasks if t["id"] == task_id), None)

¿Por qué estas decisiones?

• Lista de dicts: Simula una DB real — retorna datos "crudos" que los endpoints convierten a Pydantic.
• id_counter global: max(t["id"]...) + 1 fallaría si borras la tarea con ID más alto. El contador siempre avanza.
• find_task retorna None: El endpoint decide qué hacer (404, etc.), no el helper.
• 6 tareas variadas: ?status=pending → 3 resultados, ?priority=high → 2, combinados → 1.

Probando tus modelos

No necesitas levantar el servidor. Prueba directamente desde la terminal (desde todo-api/ con venv activado).

Tests de creación y defaults

python -c "
from app.models import TaskCreate

# Test 1: Crear con solo title — verifica defaults
task = TaskCreate(title='Mi primera tarea')
print('Válida:', task.model_dump())
# → description: None, status: 'pending', priority: 'medium'

# Test 2: Título vacío — debe fallar (min_length=1)
try:
    TaskCreate(title='')
except Exception as e:
    print(f'Título vacío: Error esperado')

# Test 3: Título solo espacios — debe fallar (validator)
try:
    TaskCreate(title='   ')
except Exception as e:
    print(f'Solo espacios: Error esperado')
"

Tests de validación de campos

python -c "
from app.models import TaskCreate

# Status inválido — debe fallar (Literal)
try:
    TaskCreate(title='Test', status='invalid')
except Exception as e:
    print(f'Status inválido: Error esperado')

# Descripción demasiado larga — debe fallar (max_length=1000)
try:
    TaskCreate(title='Test', description='x' * 1001)
except Exception as e:
    print(f'Desc larga: Error esperado')
"

Test de exclude_unset (clave para PATCH)

python -c "
from app.models import TaskPatch

patch = TaskPatch(status='completed')
print('model_dump():', patch.model_dump())
print('exclude_unset:', patch.model_dump(exclude_unset=True))
"

Esperado:

model_dump(): {'title': None, 'description': None, 'status': 'completed', 'priority': None}
exclude_unset: {'status': 'completed'}

Sin exclude_unset=True, PATCH sobreescribiría todo con None.

Verificar datos iniciales

python -c "
from app.data import tasks, find_task, generate_id, get_current_timestamp

print(f'Tareas: {len(tasks)}')
print(f'Tarea 3: {find_task(3)[\"title\"]}')
print(f'Tarea 99: {find_task(99)}')
print(f'Siguiente ID: {generate_id()}')
print(f'Timestamp: {get_current_timestamp()}')
"

Ejercicios

Ejercicio 1: Herencia de validators (Fácil)

Crea un TaskCreate con title=" Mi tarea con espacios ". Imprime task.title. ¿Se aplicó el strip() del validator heredado de TaskBase?

from app.models import TaskCreate

task = TaskCreate(title="  Mi tarea con espacios  ")
print(repr(task.title))
# → 'Mi tarea con espacios'

Ejercicio 2: TaskUpdate requiere todo (Medio)

Intenta crear un TaskUpdate sin enviar status. ¿Qué error obtienes? ¿Y si envías solo title?

from app.models import TaskUpdate

try:
    TaskUpdate(title="Test", description=None, priority="high")
except Exception as e:
    print(f"Sin status: {e}")
# → Field required: status

try:
    TaskUpdate(title="Test")
except Exception as e:
    print(f"Solo title: {e}")
# → Field required: description, status, priority

Ejercicio 3: Flujo crear + patch (Difícil)

Sin levantar el servidor, simula: crear una tarea con TaskCreate, generar id y created_at, hacer PATCH para cambiar solo el status, y validar el resultado con TaskResponse.

from app.models import TaskCreate, TaskPatch, TaskResponse
from app.data import generate_id, get_current_timestamp

create_data = TaskCreate(title="Tarea de prueba", priority="high")
now = get_current_timestamp()
task_dict = {
    "id": generate_id(),
    **create_data.model_dump(),
    "created_at": now,
    "updated_at": None,
}
print("Creada:", task_dict)

patch_data = TaskPatch(status="completed")
update_fields = patch_data.model_dump(exclude_unset=True)
print("Campos a actualizar:", update_fields)

task_dict.update(update_fields)
task_dict["updated_at"] = get_current_timestamp()
print("Después del PATCH:", task_dict)

response = TaskResponse(**task_dict)
print("Response válida:", response.model_dump())

Checklist de verificación

Antes de pasar a la siguiente cápsula, verifica cada punto:

✅ app/models.py
├── [ ] Imports: Literal, BaseModel, Field, field_validator
├── [ ] ALLOWED_STATUSES: pending, in_progress, completed
├── [ ] ALLOWED_PRIORITIES: low, medium, high, urgent
├── [ ] TaskBase: 4 campos con Field() constraints
├── [ ] TaskBase: validator para title (strip + whitespace check)
├── [ ] TaskBase: validator para description (strip + None)
├── [ ] TaskCreate hereda de TaskBase (pass)
├── [ ] TaskUpdate hereda de TaskBase con campos requeridos
├── [ ] TaskPatch independiente, todo Optional, con validators propios
├── [ ] TaskResponse hereda de TaskBase + id, created_at, updated_at
├── [ ] TaskListResponse con tasks y count
└── [ ] Sin errores: python -c "from app.models import *"

✅ app/data.py
├── [ ] tasks tiene 6 dicts con todos los campos
├── [ ] Combinaciones variadas de status y priority
├── [ ] generate_id() retorna IDs incrementales
├── [ ] get_current_timestamp() retorna string ISO 8601
├── [ ] find_task() retorna dict o None
└── [ ] Sin errores: python -c "from app.data import *"

✅ Tests manuales
├── [ ] TaskCreate con solo title → defaults correctos
├── [ ] TaskCreate con title vacío → falla
├── [ ] TaskCreate con status inválido → falla
├── [ ] TaskPatch.model_dump(exclude_unset=True) → solo campos enviados
├── [ ] find_task(3) → retorna dict
└── [ ] find_task(99) → retorna None

Estado del proyecto

✅ Completado:
├── Estructura de carpetas profesional (cápsula 02)
├── Modelos: TaskBase, TaskCreate, TaskUpdate, TaskPatch, TaskResponse, TaskListResponse
├── Validators: title whitespace, description cleanup
├── Literal types para status y priority
├── 6 tareas iniciales con datos realistas
├── Helpers: generate_id(), get_current_timestamp(), find_task()
└── Modelos verificados con tests manuales

⬜ Pendiente:
├── Endpoints CRUD (cápsula 04)
├── Error handling + CORS + docs (cápsula 05)
└── Verificación final y entrega (cápsula 06)

Resumen

• TaskBase centraliza campos y validación — los cambios se propagan a TaskCreate, TaskUpdate y TaskResponse
• TaskCreate hereda todo de TaskBase sin agregar nada — el servidor genera id y timestamps
• TaskUpdate redefine campos con Field(...) para hacerlos requeridos — PUT reemplaza completo
• TaskPatch es independiente con todo Optional — PATCH solo actualiza lo enviado
• TaskResponse agrega id, created_at, updated_at — es lo que el cliente recibe
• Literal restringe status y priority a valores específicos sin Enums
• @field_validator maneja validaciones que Field() no cubre: whitespace-only, strip automático
• model_dump(exclude_unset=True) es la clave para que PATCH funcione
• El almacenamiento in-memory con list[dict] simula una base de datos
• find_task retorna None — el endpoint decide cómo reaccionar
• Las 6 tareas cubren diferentes combinaciones de status y priority para facilitar pruebas

Recursos adicionales

1. Pydantic v2 — BaseModel — Referencia de modelos y herencia
2. Pydantic v2 — Field — Constraints, defaults, description y examples
3. Pydantic v2 — Validators — @field_validator, @model_validator y modos
4. Pydantic v2 — Serialization — model_dump(), exclude_unset, exclude_none
5. Python — typing.Literal — Restringir valores sin Enums
6. FastAPI — Request Body — Cómo FastAPI usa Pydantic models como request body

Siguiente cápsula: Endpoints CRUD — Implementarás POST, GET, PUT, PATCH y DELETE, conectando los modelos que definiste aquí con las rutas de FastAPI, incluyendo filtros y paginación.