Modelos SQLAlchemy y Pydantic schemas

Diferencia entre modelos SQLAlchemy y schemas Pydantic

En FastAPI trabajamos con dos tipos de estructuras de datos que, aunque pueden parecer similares, tienen propósitos completamente diferentes. Esta distinción es fundamental para construir aplicaciones web robustas y bien organizadas.

Los modelos SQLAlchemy representan las tablas de nuestra base de datos. Son clases que definen la estructura de los datos tal como se almacenan en la base de datos, incluyendo tipos de columnas, restricciones y relaciones. Por otro lado, los schemas Pydantic definen la estructura de los datos que viajan a través de nuestra API, tanto en las peticiones que recibimos como en las respuestas que enviamos.

Modelos SQLAlchemy: la representación de la base de datos

Un modelo SQLAlchemy es una clase Python que hereda de Base y representa una tabla en la base de datos. Cada atributo de la clase corresponde a una columna de la tabla:

from sqlalchemy import Column, Integer, String, Boolean
from database import Base

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    email = Column(String, unique=True, nullable=False)
    is_active = Column(Boolean, default=True)

Este modelo define exactamente cómo se estructura la tabla users en la base de datos. SQLAlchemy utiliza esta información para crear las tablas, realizar consultas y gestionar las operaciones de base de datos.

Schemas Pydantic: la interfaz de la API

Los schemas Pydantic son clases que heredan de BaseModel y definen la estructura de los datos que nuestra API acepta y devuelve:

from pydantic import BaseModel
from typing import Optional

class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool
    
    class Config:
        from_attributes = True

El schema UserCreate define qué datos necesitamos para crear un usuario, mientras que UserResponse especifica qué información devolvemos cuando consultamos un usuario. Pydantic se encarga de validar automáticamente que los datos cumplan con estos esquemas.

Diferencias clave en la práctica

La separación de responsabilidades es la diferencia más importante. Los modelos SQLAlchemy se enfocan en la persistencia de datos, mientras que los schemas Pydantic se centran en la validación y serialización para la API.

Consideremos un ejemplo práctico. Nuestro modelo SQLAlchemy incluye campos internos que no queremos exponer:

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    email = Column(String, unique=True, nullable=False)
    password_hash = Column(String, nullable=False)  # Campo interno
    created_at = Column(DateTime, default=datetime.utcnow)
    is_active = Column(Boolean, default=True)

Sin embargo, nuestro schema de respuesta solo expone la información segura y relevante:

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool
    # No incluimos password_hash por seguridad
    
    class Config:
        from_attributes = True

Flexibilidad en los schemas

Los schemas Pydantic nos permiten tener múltiples representaciones del mismo modelo según el contexto. Para el mismo modelo User, podemos tener diferentes schemas:

# Para crear un usuario (sin ID, será generado automáticamente)
class UserCreate(BaseModel):
    name: str
    email: str
    password: str

# Para actualizar un usuario (campos opcionales)
class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None

# Para la respuesta pública (sin datos sensibles)
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool
    
    class Config:
        from_attributes = True

Esta flexibilidad nos permite adaptar la interfaz de nuestra API a diferentes necesidades sin modificar la estructura de la base de datos.

Validación automática

Pydantic proporciona validación automática de tipos y formatos, algo que SQLAlchemy no hace por sí solo en el contexto de una API:

from pydantic import BaseModel, EmailStr
from typing import Optional

class UserCreate(BaseModel):
    name: str
    email: EmailStr  # Valida formato de email automáticamente
    age: Optional[int] = None
    
    # Pydantic validará que name sea string, email tenga formato válido
    # y age sea entero o None

Cuando FastAPI recibe una petición, Pydantic valida automáticamente que los datos cumplan con el schema antes de que lleguen a nuestra función. Si no cumplen, devuelve un error 422 con detalles específicos del problema.

Esta separación clara entre modelos y schemas nos permite mantener una arquitectura limpia donde cada componente tiene una responsabilidad específica: SQLAlchemy maneja la persistencia y Pydantic gestiona la validación y serialización de la API.

Conversión entre modelos y schemas

Una vez que comprendemos la diferencia entre modelos SQLAlchemy y schemas Pydantic, necesitamos aprender cómo convertir datos entre estas dos representaciones. Esta conversión es esencial en cualquier aplicación FastAPI que trabaje con bases de datos.

Conversión automática con from_attributes

Pydantic puede convertir automáticamente objetos SQLAlchemy a schemas utilizando la configuración from_attributes = True. Esta característica permite que Pydantic acceda a los atributos del modelo SQLAlchemy como si fueran un diccionario:

# Modelo SQLAlchemy
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    email = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)

# Schema Pydantic con conversión automática
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool
    
    class Config:
        from_attributes = True

Con esta configuración, podemos convertir directamente un objeto SQLAlchemy a un schema Pydantic:

# Obtener usuario de la base de datos
db_user = session.query(User).first()

# Conversión automática a schema
user_response = UserResponse.from_orm(db_user)
# O en versiones más recientes de Pydantic:
user_response = UserResponse.model_validate(db_user)

Conversión en endpoints de FastAPI

En la práctica, esta conversión ocurre automáticamente cuando especificamos el tipo de respuesta en nuestros endpoints:

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
    # Consulta que devuelve un modelo SQLAlchemy
    db_user = db.query(User).filter(User.id == user_id).first()
    
    if not db_user:
        raise HTTPException(status_code=404, detail="Usuario no encontrado")
    
    # FastAPI convierte automáticamente db_user a UserResponse
    return db_user

FastAPI detecta que el tipo de retorno es UserResponse y automáticamente convierte el objeto SQLAlchemy utilizando Pydantic. No necesitamos hacer la conversión manualmente.

Conversión de schema a modelo

Para crear nuevos registros, necesitamos convertir los datos del schema Pydantic a un modelo SQLAlchemy. Esto se hace extrayendo los datos del schema y pasándolos al constructor del modelo:

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # Convertir schema a diccionario
    user_data = user.model_dump()
    
    # Crear modelo SQLAlchemy con los datos
    db_user = User(**user_data)
    
    # Guardar en base de datos
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    
    # FastAPI convierte automáticamente a UserResponse
    return db_user

El método model_dump() extrae todos los datos del schema Pydantic como un diccionario Python, que luego podemos usar para crear el modelo SQLAlchemy.

Manejo de campos adicionales

Cuando el schema tiene campos diferentes al modelo, necesitamos manejar la conversión de forma más específica:

class UserCreate(BaseModel):
    name: str
    email: str
    password: str  # Campo que no existe en el modelo final

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    email = Column(String, nullable=False)
    password_hash = Column(String, nullable=False)  # Campo transformado
    is_active = Column(Boolean, default=True)

En este caso, necesitamos procesar los datos antes de crear el modelo:

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # Extraer datos del schema
    user_data = user.model_dump()
    
    # Procesar campo especial
    password = user_data.pop("password")  # Extraer password
    password_hash = hash_password(password)  # Función para hashear
    
    # Crear modelo con datos procesados
    db_user = User(
        **user_data,
        password_hash=password_hash
    )
    
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    
    return db_user

Conversión para actualizaciones

Para actualizar registros existentes, combinamos datos del schema con el modelo existente:

@app.put("/users/{user_id}", response_model=UserResponse)
def update_user(user_id: int, user_update: UserUpdate, db: Session = Depends(get_db)):
    # Obtener modelo existente
    db_user = db.query(User).filter(User.id == user_id).first()
    
    if not db_user:
        raise HTTPException(status_code=404, detail="Usuario no encontrado")
    
    # Obtener solo campos con valores (exclude_unset=True)
    update_data = user_update.model_dump(exclude_unset=True)
    
    # Actualizar campos del modelo
    for field, value in update_data.items():
        setattr(db_user, field, value)
    
    db.commit()
    db.refresh(db_user)
    
    return db_user

El parámetro exclude_unset=True en model_dump() nos permite obtener solo los campos que fueron proporcionados en la petición, ignorando los campos con valores por defecto.

Conversión de listas

Cuando trabajamos con múltiples registros, la conversión sigue el mismo patrón:

@app.get("/users/", response_model=List[UserResponse])
def get_users(db: Session = Depends(get_db)):
    # Consulta que devuelve lista de modelos SQLAlchemy
    db_users = db.query(User).all()
    
    # FastAPI convierte automáticamente cada elemento
    return db_users

FastAPI aplica la conversión a cada elemento de la lista automáticamente, transformando todos los modelos SQLAlchemy a schemas Pydantic.

Esta conversión bidireccional entre modelos y schemas es lo que permite que FastAPI mantenga una separación clara entre la lógica de base de datos y la interfaz de la API, mientras proporciona validación automática y documentación precisa.