Tutorial Flask: Serialización Pydantic

Modelos Pydantic en Flask

Pydantic es una biblioteca de validación de datos que utiliza anotaciones de tipo de Python para definir esquemas de datos robustos. A diferencia de Marshmallow, que hemos visto anteriormente, Pydantic aprovecha las características nativas de Python para crear modelos que validan automáticamente los datos de entrada y salida.

La principal ventaja de Pydantic radica en su integración natural con el sistema de tipos de Python. Esto significa que obtienes validación automática, serialización y documentación de tu API sin escribir código adicional complejo.

Instalación y configuración básica

Para comenzar a trabajar con Pydantic en Flask, necesitas instalar la biblioteca:

pip install pydantic

Una vez instalado, puedes crear tu primer modelo Pydantic. Los modelos se definen como clases que heredan de BaseModel:

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class Usuario(BaseModel):
    id: Optional[int] = None
    nombre: str
    email: str
    edad: int
    activo: bool = True
    fecha_registro: Optional[datetime] = None

Este modelo define la estructura de un usuario con validación automática de tipos. Pydantic verificará que nombre y email sean cadenas, edad sea un entero, y activo sea un booleano.

Integración con rutas Flask

Para utilizar modelos Pydantic en tus endpoints de Flask, necesitas extraer los datos JSON de la petición y crear una instancia del modelo:

from flask import Flask, request, jsonify
from pydantic import ValidationError

app = Flask(__name__)

# Simulamos una base de datos en memoria
usuarios_db = []
contador_id = 1

@app.route('/usuarios', methods=['POST'])
def crear_usuario():
    global contador_id
    
    try:
        # Crear instancia del modelo con los datos recibidos
        usuario_data = Usuario(**request.json)
        
        # Asignar ID y fecha de registro
        usuario_data.id = contador_id
        usuario_data.fecha_registro = datetime.now()
        
        # Guardar en nuestra "base de datos"
        usuarios_db.append(usuario_data.model_dump())
        contador_id += 1
        
        return jsonify(usuario_data.model_dump()), 201
        
    except ValidationError as e:
        return jsonify({"errores": e.errors()}), 400

El método model_dump() convierte el modelo Pydantic en un diccionario Python que puede ser serializado a JSON. Si los datos no cumplen con las validaciones, Pydantic lanza una ValidationError con detalles específicos sobre qué campos fallaron.

Validaciones avanzadas con Pydantic

Pydantic ofrece validadores personalizados que van más allá de la simple verificación de tipos. Puedes definir reglas de negocio específicas usando decoradores:

from pydantic import BaseModel, validator, Field
import re

class Producto(BaseModel):
    nombre: str = Field(..., min_length=3, max_length=100)
    precio: float = Field(..., gt=0, description="El precio debe ser mayor que 0")
    categoria: str
    codigo_sku: str
    
    @validator('codigo_sku')
    def validar_sku(cls, v):
        if not re.match(r'^[A-Z]{2}\d{4}$', v):
            raise ValueError('El SKU debe tener formato XX0000 (2 letras + 4 números)')
        return v
    
    @validator('categoria')
    def validar_categoria(cls, v):
        categorias_validas = ['electronica', 'ropa', 'hogar', 'deportes']
        if v.lower() not in categorias_validas:
            raise ValueError(f'Categoría debe ser una de: {categorias_validas}')
        return v.lower()

Modelos anidados y relaciones

Una característica destacada de Pydantic es su capacidad para manejar estructuras de datos complejas mediante modelos anidados:

class Direccion(BaseModel):
    calle: str
    ciudad: str
    codigo_postal: str
    pais: str = "España"

class UsuarioCompleto(BaseModel):
    id: Optional[int] = None
    nombre: str
    email: str
    direccion: Direccion
    telefonos: list[str] = []
    
    @validator('telefonos')
    def validar_telefonos(cls, v):
        for telefono in v:
            if not re.match(r'^\+?[\d\s-]{9,15}$', telefono):
                raise ValueError(f'Teléfono inválido: {telefono}')
        return v

Con este modelo, puedes recibir datos JSON complejos y Pydantic validará automáticamente tanto el objeto principal como los objetos anidados:

@app.route('/usuarios-completos', methods=['POST'])
def crear_usuario_completo():
    try:
        usuario = UsuarioCompleto(**request.json)
        
        # El modelo ya está validado completamente
        usuarios_db.append(usuario.model_dump())
        
        return jsonify(usuario.model_dump()), 201
        
    except ValidationError as e:
        return jsonify({"errores": e.errors()}), 400

Configuración de modelos

Pydantic permite personalizar el comportamiento de los modelos mediante la clase Config. Esto es especialmente útil para controlar cómo se manejan los datos de entrada y salida:

class UsuarioConfigurable(BaseModel):
    nombre: str
    email: str
    password: str
    
    class Config:
        # Permitir campos adicionales sin error
        extra = "forbid"
        # Validar asignaciones después de la creación
        validate_assignment = True
        # Usar enums por valor
        use_enum_values = True
        # Personalizar nombres de campos en JSON
        fields = {
            "password": {"write_only": True}
        }
    
    def model_dump_public(self):
        """Método personalizado para excluir campos sensibles"""
        return self.model_dump(exclude={"password"})

Esta configuración hace que el modelo sea más estricto rechazando campos no definidos y validando cambios posteriores a la creación del objeto.

Serialización automática

La serialización automática en Pydantic elimina la necesidad de escribir código manual para convertir objetos Python en JSON y viceversa. Esta funcionalidad se integra perfectamente con Flask para crear APIs que manejan datos de forma transparente y eficiente.

Serialización de respuestas con model_dump()

El método model_dump() es la herramienta principal para convertir modelos Pydantic en diccionarios serializables. Este método ofrece múltiples opciones de configuración para controlar exactamente qué datos se incluyen en la respuesta:

from pydantic import BaseModel
from datetime import datetime
from typing import Optional

class Pedido(BaseModel):
    id: int
    cliente_id: int
    productos: list[str]
    total: float
    fecha_creacion: datetime
    notas_internas: Optional[str] = None
    procesado: bool = False

# Datos de ejemplo
pedido = Pedido(
    id=1,
    cliente_id=123,
    productos=["Laptop", "Mouse"],
    total=899.99,
    fecha_creacion=datetime.now(),
    notas_internas="Cliente VIP",
    procesado=True
)

# Serialización básica
pedido_json = pedido.model_dump()

Exclusión selectiva de campos

Una característica fundamental de la serialización automática es la capacidad de excluir campos específicos según el contexto. Esto es especialmente útil para APIs públicas donde cierta información debe mantenerse privada:

@app.route('/pedidos/<int:pedido_id>')
def obtener_pedido(pedido_id):
    pedido = encontrar_pedido(pedido_id)
    
    if not pedido:
        return jsonify({"error": "Pedido no encontrado"}), 404
    
    # Excluir información sensible para usuarios normales
    pedido_publico = pedido.model_dump(exclude={"notas_internas"})
    
    return jsonify(pedido_publico)

@app.route('/admin/pedidos/<int:pedido_id>')
def obtener_pedido_admin(pedido_id):
    pedido = encontrar_pedido(pedido_id)
    
    if not pedido:
        return jsonify({"error": "Pedido no encontrado"}), 404
    
    # Los administradores ven toda la información
    return jsonify(pedido.model_dump())

Inclusión específica de campos

El parámetro include permite seleccionar únicamente los campos que deseas serializar, útil para crear vistas ligeras de tus datos:

@app.route('/pedidos/resumen')
def resumen_pedidos():
    pedidos = obtener_todos_los_pedidos()
    
    # Solo incluir campos esenciales para el resumen
    resumen = [
        pedido.model_dump(include={"id", "total", "fecha_creacion", "procesado"})
        for pedido in pedidos
    ]
    
    return jsonify(resumen)

Serialización con alias de campos

Pydantic permite definir nombres alternativos para los campos durante la serialización, lo que facilita la integración con sistemas externos que esperan nomenclaturas específicas:

from pydantic import BaseModel, Field

class ProductoAPI(BaseModel):
    identificador: int = Field(alias="product_id")
    nombre_producto: str = Field(alias="productName")
    precio_unitario: float = Field(alias="unitPrice")
    disponible: bool = Field(alias="inStock")
    
    class Config:
        populate_by_name = True  # Permite usar tanto el nombre original como el alias

@app.route('/productos/<int:producto_id>')
def obtener_producto(producto_id):
    producto = ProductoAPI(
        identificador=producto_id,
        nombre_producto="Teclado mecánico",
        precio_unitario=89.99,
        disponible=True
    )
    
    # Serializa usando los alias definidos
    return jsonify(producto.model_dump(by_alias=True))

Manejo automático de tipos complejos

La serialización automática de Pydantic maneja tipos de datos complejos sin configuración adicional, incluyendo fechas, enums y objetos anidados:

from enum import Enum
from datetime import datetime, date

class EstadoPedido(Enum):
    PENDIENTE = "pendiente"
    PROCESANDO = "procesando"
    ENVIADO = "enviado"
    ENTREGADO = "entregado"

class PedidoComplejo(BaseModel):
    id: int
    estado: EstadoPedido
    fecha_pedido: datetime
    fecha_entrega_estimada: date
    metadatos: dict[str, any] = {}
    
    class Config:
        use_enum_values = True  # Serializa enums por su valor

@app.route('/pedidos-complejos', methods=['POST'])
def crear_pedido_complejo():
    datos = request.json
    
    try:
        pedido = PedidoComplejo(**datos)
        
        # La serialización maneja automáticamente:
        # - Enum -> string
        # - datetime -> ISO format string
        # - date -> ISO format string
        # - dict -> JSON object
        
        return jsonify(pedido.model_dump()), 201
        
    except ValidationError as e:
        return jsonify({"errores": e.errors()}), 400

Serialización condicional con model_dump_json()

Para casos donde necesitas control granular sobre el formato JSON, Pydantic ofrece model_dump_json() que serializa directamente a string JSON con opciones avanzadas:

@app.route('/pedidos/<int:pedido_id>/export')
def exportar_pedido(pedido_id):
    pedido = encontrar_pedido(pedido_id)
    
    if not pedido:
        return jsonify({"error": "Pedido no encontrado"}), 404
    
    # Serialización directa a JSON con formato personalizado
    json_string = pedido.model_dump_json(
        exclude_none=True,  # Omite campos con valor None
        indent=2,           # Formato legible
        ensure_ascii=False  # Permite caracteres Unicode
    )
    
    return json_string, 200, {'Content-Type': 'application/json; charset=utf-8'}

Serialización de listas y colecciones

Cuando trabajas con colecciones de modelos, la serialización automática se aplica a cada elemento de forma transparente:

@app.route('/usuarios/activos')
def usuarios_activos():
    usuarios = [
        Usuario(nombre="Ana", email="ana@email.com", edad=28, activo=True),
        Usuario(nombre="Carlos", email="carlos@email.com", edad=35, activo=True),
        Usuario(nombre="María", email="maria@email.com", edad=42, activo=True)
    ]
    
    # Serialización automática de toda la lista
    usuarios_serializados = [usuario.model_dump() for usuario in usuarios]
    
    return jsonify({
        "total": len(usuarios_serializados),
        "usuarios": usuarios_serializados
    })

Personalización de la serialización

Para casos específicos donde necesitas lógica de serialización personalizada, puedes sobrescribir métodos o usar validadores de serialización:

class UsuarioPersonalizado(BaseModel):
    nombre: str
    email: str
    fecha_nacimiento: date
    
    def model_dump_publico(self):
        """Método personalizado para vista pública"""
        data = self.model_dump()
        
        # Calcular edad dinámicamente
        hoy = date.today()
        edad = hoy.year - self.fecha_nacimiento.year
        
        # Ocultar información sensible y agregar campos calculados
        return {
            "nombre": data["nombre"],
            "edad": edad,
            "email_dominio": data["email"].split("@")[1]
        }

@app.route('/usuarios/<int:user_id>/publico')
def perfil_publico(user_id):
    usuario = encontrar_usuario(user_id)
    
    if not usuario:
        return jsonify({"error": "Usuario no encontrado"}), 404
    
    return jsonify(usuario.model_dump_publico())