DTOs y Mappers

Data Transfer Objects

Los Data Transfer Objects (DTOs) son objetos que transportan datos entre diferentes capas de una aplicación, actuando como un puente entre la capa de presentación y la lógica de negocio. En el contexto de FastAPI, los DTOs nos permiten separar claramente lo que exponemos en nuestra API de la estructura interna de nuestros modelos de base de datos.

La necesidad de DTOs surge cuando queremos controlar exactamente qué información se envía y recibe a través de nuestra API, sin estar limitados por la estructura de nuestros modelos SQLAlchemy. Esto nos proporciona flexibilidad, seguridad y una mejor arquitectura.

¿Por qué usar DTOs?

Imagina que tienes un modelo User en SQLAlchemy que contiene información sensible como contraseñas hash, tokens de sesión, o datos internos que no deberían ser expuestos públicamente. Los DTOs nos permiten filtrar y transformar esta información antes de enviarla al cliente.

from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, nullable=False)
    password_hash = Column(String, nullable=False)  # Información sensible
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime)
    last_login = Column(DateTime)
    session_token = Column(String)  # Información interna

Para este modelo, no queremos exponer el password_hash ni el session_token en nuestras respuestas API. Aquí es donde entran los DTOs.

Creando DTOs con Pydantic

En FastAPI, los DTOs se implementan típicamente usando Pydantic models. Estos modelos definen exactamente qué campos queremos incluir en nuestras transferencias de datos y sus tipos correspondientes.

from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional

class UserResponseDTO(BaseModel):
    id: int
    email: EmailStr
    is_active: bool
    created_at: datetime
    last_login: Optional[datetime] = None
    
    class Config:
        from_attributes = True

El UserResponseDTO incluye solo los campos que queremos exponer públicamente, omitiendo información sensible como contraseñas y tokens.

DTOs para diferentes operaciones

Una práctica común es crear DTOs específicos para diferentes operaciones CRUD, cada uno adaptado a su propósito específico.

DTO para creación de usuarios:

class UserCreateDTO(BaseModel):
    email: EmailStr
    password: str  # Contraseña en texto plano que será procesada
    
    class Config:
        # Ejemplo de validación personalizada
        schema_extra = {
            "example": {
                "email": "usuario@ejemplo.com",
                "password": "mi_contraseña_segura"
            }
        }

DTO para actualización de usuarios:

class UserUpdateDTO(BaseModel):
    email: Optional[EmailStr] = None
    password: Optional[str] = None
    is_active: Optional[bool] = None

DTO para respuestas de listado:

class UserListItemDTO(BaseModel):
    id: int
    email: EmailStr
    is_active: bool
    
    class Config:
        from_attributes = True

Validación y transformación en DTOs

Los DTOs no solo transportan datos, también pueden validar y transformar información antes de procesarla. Pydantic nos permite agregar validadores personalizados que se ejecutan automáticamente.

from pydantic import field_validator
import re

class UserCreateDTO(BaseModel):
    email: EmailStr
    password: str
    
    @field_validator('password')
    def validate_password_strength(cls, password, info):
        if len(password) < 8:
            raise ValueError('La contraseña debe tener al menos 8 caracteres')
        
        if not re.search(r'[A-Z]', password):
            raise ValueError('La contraseña debe contener al menos una mayúscula')
            
        if not re.search(r'[0-9]', password):
            raise ValueError('La contraseña debe contener al menos un número')
            
        return password

DTOs anidados para relaciones

Cuando trabajamos con modelos que tienen relaciones, podemos crear DTOs anidados que representen estas asociaciones de manera controlada.

class PostSummaryDTO(BaseModel):
    id: int
    title: str
    created_at: datetime
    
    class Config:
        from_attributes = True

class UserWithPostsDTO(BaseModel):
    id: int
    email: EmailStr
    is_active: bool
    posts: List[PostSummaryDTO] = []
    
    class Config:
        from_attributes = True

Este enfoque nos permite controlar la profundidad de los datos relacionados que incluimos en nuestras respuestas, evitando problemas como la serialización circular o la exposición de demasiada información.

Usando DTOs en endpoints

Los DTOs se integran naturalmente en los endpoints de FastAPI, proporcionando validación automática de entrada y garantizando la consistencia de las respuestas.

from fastapi import FastAPI, HTTPException
from typing import List

app = FastAPI()

@app.post("/users", response_model=UserResponseDTO)
async def create_user(user_data: UserCreateDTO):
    # La validación del DTO ocurre automáticamente
    # user_data.password ya está validado según nuestras reglas
    
    # Aquí iría la lógica de creación del usuario
    # (que veremos en la siguiente sección sobre mapeo)
    pass

@app.get("/users", response_model=List[UserListItemDTO])
async def list_users():
    # Devolver lista de usuarios usando el DTO de listado
    pass

@app.get("/users/{user_id}", response_model=UserResponseDTO)
async def get_user(user_id: int):
    # Devolver un usuario específico
    pass

Beneficios de los DTOs

El uso de DTOs en FastAPI nos proporciona múltiples ventajas:

• Seguridad: Control total sobre qué información se expone
• Flexibilidad: Diferentes representaciones para diferentes casos de uso
• Validación: Reglas de negocio aplicadas automáticamente
• Documentación: FastAPI genera automáticamente la documentación OpenAPI basada en los DTOs
• Evolución: Capacidad de cambiar la API sin afectar los modelos internos

Los DTOs actúan como un contrato bien definido entre nuestra API y los clientes que la consumen, proporcionando estabilidad y claridad en las comunicaciones. En la siguiente sección veremos cómo transformar eficientemente entre nuestros modelos SQLAlchemy y estos DTOs usando técnicas de mapeo.

Mapeo entre modelos

El mapeo entre modelos es el proceso de convertir datos entre diferentes representaciones, específicamente entre nuestros modelos SQLAlchemy y los DTOs que hemos definido. Esta transformación es esencial para mantener la separación entre la capa de datos y la capa de presentación en nuestras aplicaciones FastAPI.

La necesidad del mapeo surge porque los modelos SQLAlchemy están diseñados para representar la estructura de la base de datos, mientras que los DTOs están optimizados para la comunicación a través de la API. Cada uno tiene un propósito específico y requiere estructuras diferentes.

Mapeo básico modelo a DTO

La forma más directa de mapear un modelo SQLAlchemy a un DTO es utilizar la configuración from_attributes = True en Pydantic, que permite crear instancias del DTO directamente desde objetos con atributos.

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

# DTO
class UserResponseDTO(BaseModel):
    id: int
    email: str
    full_name: str
    is_active: bool
    created_at: datetime
    
    class Config:
        from_attributes = True

# Mapeo directo
def map_user_to_dto(user: User) -> UserResponseDTO:
    return UserResponseDTO.model_validate(user)

Este enfoque funciona cuando los nombres de los campos coinciden entre el modelo y el DTO. Pydantic automáticamente extrae los valores de los atributos del objeto SQLAlchemy.

Funciones mapper personalizadas

Cuando necesitamos más control sobre la transformación, podemos crear funciones mapper personalizadas que manejen la conversión de manera explícita.

def map_user_to_response_dto(user: User) -> UserResponseDTO:
    """
    Convierte un modelo User de SQLAlchemy a UserResponseDTO.
    Incluye transformaciones personalizadas si son necesarias.
    """
    return UserResponseDTO(
        id=user.id,
        email=user.email,
        full_name=user.full_name,
        is_active=user.is_active,
        created_at=user.created_at
    )

def map_user_to_list_item_dto(user: User) -> UserListItemDTO:
    """
    Convierte un modelo User a una versión resumida para listados.
    """
    return UserListItemDTO(
        id=user.id,
        email=user.email,
        is_active=user.is_active
    )

Mapeo DTO a modelo

Para el mapeo inverso, desde DTO hacia modelo SQLAlchemy, necesitamos ser más explícitos ya que estamos creando instancias de clases que interactuarán con la base de datos.

def map_create_dto_to_user(user_dto: UserCreateDTO) -> User:
    """
    Convierte un UserCreateDTO en un modelo User de SQLAlchemy.
    Excluye campos que se generan automáticamente como ID y timestamps.
    """
    return User(
        email=user_dto.email,
        full_name=user_dto.full_name,
        # password_hash se calcularía aquí usando un hash seguro
        # created_at se asigna automáticamente por el modelo
    )

def update_user_from_dto(user: User, update_dto: UserUpdateDTO) -> User:
    """
    Actualiza un modelo User existente con datos del DTO de actualización.
    Solo modifica los campos que están presentes en el DTO.
    """
    if update_dto.email is not None:
        user.email = update_dto.email
    
    if update_dto.full_name is not None:
        user.full_name = update_dto.full_name
        
    if update_dto.is_active is not None:
        user.is_active = update_dto.is_active
        
    return user

Mapeo con transformaciones de datos

A menudo necesitamos aplicar transformaciones durante el mapeo, como calcular campos derivados, formatear datos o aplicar lógica de negocio específica.

from datetime import datetime, timezone

def map_user_with_computed_fields(user: User) -> UserResponseDTO:
    """
    Mapea un usuario incluyendo campos calculados y transformaciones.
    """
    # Calcular tiempo desde la creación
    time_since_creation = datetime.utcnow() - user.created_at
    days_active = time_since_creation.days
    
    return UserResponseDTO(
        id=user.id,
        email=user.email,
        full_name=user.full_name.title(),  # Formatear nombre
        is_active=user.is_active,
        created_at=user.created_at,
        days_since_registration=days_active  # Campo calculado
    )

Mapeo de listas y colecciones

Cuando trabajamos con múltiples registros, necesitamos mapear colecciones de manera eficiente. Python nos proporciona varias herramientas para esto.

from typing import List

def map_users_to_list_dto(users: List[User]) -> List[UserListItemDTO]:
    """
    Convierte una lista de usuarios a DTOs de listado.
    """
    return [map_user_to_list_item_dto(user) for user in users]

# Alternativa usando map()
def map_users_using_builtin_map(users: List[User]) -> List[UserListItemDTO]:
    """
    Utiliza la función map() incorporada para mayor eficiencia.
    """
    return list(map(map_user_to_list_item_dto, users))

Mapeo de relaciones complejas

Cuando nuestros modelos tienen relaciones con otros modelos, el mapeo se vuelve más complejo. Necesitamos decidir qué nivel de profundidad incluir y cómo representar estas relaciones.

# Modelos con relación
class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    content = Column(Text)
    user_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relación
    author = relationship("User", back_populates="posts")

# DTOs para la relación
class PostSummaryDTO(BaseModel):
    id: int
    title: str
    created_at: datetime
    
    class Config:
        from_attributes = True

class UserWithPostsDTO(BaseModel):
    id: int
    email: str
    full_name: str
    posts: List[PostSummaryDTO] = []
    
    class Config:
        from_attributes = True

# Mapper para relaciones
def map_user_with_posts(user: User) -> UserWithPostsDTO:
    """
    Mapea un usuario incluyendo sus posts relacionados.
    """
    return UserWithPostsDTO(
        id=user.id,
        email=user.email,
        full_name=user.full_name,
        posts=[PostSummaryDTO.model_validate(post) for post in user.posts]
    )

Clase Mapper para organización

Para proyectos más grandes, es útil organizar los mappers en una clase dedicada que agrupe todas las transformaciones relacionadas.

class UserMapper:
    """
    Clase que agrupa todas las operaciones de mapeo para el modelo User.
    """
    
    @staticmethod
    def to_response_dto(user: User) -> UserResponseDTO:
        return UserResponseDTO.model_validate(user)
    
    @staticmethod
    def to_list_item_dto(user: User) -> UserListItemDTO:
        return UserListItemDTO.model_validate(user)
    
    @staticmethod
    def from_create_dto(dto: UserCreateDTO) -> User:
        return User(
            email=dto.email,
            full_name=dto.full_name
        )
    
    @staticmethod
    def update_from_dto(user: User, dto: UserUpdateDTO) -> User:
        if dto.email is not None:
            user.email = dto.email
        if dto.full_name is not None:
            user.full_name = dto.full_name
        if dto.is_active is not None:
            user.is_active = dto.is_active
        return user
    
    @staticmethod
    def to_dto_list(users: List[User]) -> List[UserListItemDTO]:
        return [UserMapper.to_list_item_dto(user) for user in users]

Integración en endpoints

Los mappers se integran de forma natural en nuestros endpoints, proporcionando una capa de abstracción limpia entre los datos de la base de datos y las respuestas de la API.

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

app = FastAPI()

@app.post("/users", response_model=UserResponseDTO)
async def create_user(
    user_data: UserCreateDTO,
    db: Session = Depends(get_database_session)
):
    # Mapear DTO a modelo
    new_user = UserMapper.from_create_dto(user_data)
    
    # Guardar en base de datos
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    
    # Mapear modelo a DTO de respuesta
    return UserMapper.to_response_dto(new_user)

@app.get("/users", response_model=List[UserListItemDTO])
async def list_users(db: Session = Depends(get_database_session)):
    users = db.query(User).all()
    return UserMapper.to_dto_list(users)

@app.put("/users/{user_id}", response_model=UserResponseDTO)
async def update_user(
    user_id: int,
    update_data: UserUpdateDTO,
    db: Session = Depends(get_database_session)
):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="Usuario no encontrado")
    
    # Actualizar usando el mapper
    updated_user = UserMapper.update_from_dto(user, update_data)
    db.commit()
    
    return UserMapper.to_response_dto(updated_user)

Consideraciones de rendimiento

Al trabajar con mapeo, especialmente con grandes volúmenes de datos, es importante considerar el rendimiento:

• Mapeo por lotes: Para listas grandes, considera procesar en lotes
• Carga selectiva: Solo carga las relaciones que realmente necesitas
• Caché de mappers: Para transformaciones complejas que se repiten frecuentemente
• Validación condicional: Usa model_validate() solo cuando sea necesario

def map_users_efficiently(users: List[User], include_posts: bool = False) -> List[UserResponseDTO]:
    """
    Mapeo eficiente que permite controlar qué datos incluir.
    """
    if include_posts:
        return [UserMapper.to_response_with_posts(user) for user in users]
    else:
        return [UserMapper.to_response_dto(user) for user in users]

El mapeo entre modelos es una pieza fundamental en la arquitectura de aplicaciones FastAPI que usan SQLAlchemy. Proporciona la flexibilidad necesaria para mantener separadas las preocupaciones de datos y presentación, mientras permite evolucionar tanto la API como el modelo de datos de manera independiente.