DTOs y Mapeo
Los DTOs son la frontera entre capas; el mapeo explícito evita que el dominio se filtre hacia afuera.
Un DTO (Data Transfer Object) es un objeto de datos simple, sin comportamiento, que transporta información entre capas o sistemas. Su única responsabilidad es llevar datos; no protege invariantes, no toma decisiones y no tiene métodos de negocio. Son los mensajes que cruzan fronteras arquitectónicas.
El dominio nunca debe cruzar la frontera de la capa de aplicación hacia afuera. Si un aggregate llega directamente a un controller HTTP o a un consumidor de API, cualquier cambio interno del aggregate rompe el contrato externo. Los DTOs actúan como escudo: el dominio puede evolucionar libremente siempre que el mapeo siga correcto.
Existen al menos dos tipos de DTOs en la capa de aplicación: DTOs de entrada (Commands y sus campos, o schemas de validación en el adapter) y DTOs de salida (Read Models que devuelven los Query Handlers). El mapeo entre el aggregate y el DTO de salida es responsabilidad del Query Handler o de un mapper dedicado.
El mapeo debe ser explícito. Herramientas de mapeo automático que reflejan propiedades por nombre son peligrosas porque acoplan silenciosamente el nombre de las propiedades del dominio al contrato externo. Un mapper explícito documenta qué campos se exponen, permite transformaciones y hace visibles los cambios de contrato.
La ubicación del mapper depende de la complejidad: para casos simples, el Query Handler puede construir el Read Model directamente; para casos complejos o reutilizables, un mapper dedicado (función o clase pura) mejora la legibilidad. Lo importante es que el mapper vive en la capa de aplicación, no en el dominio.
# --- DTO de entrada (Command) ---
# application/commands/<do_something>/command.py
from dataclasses import dataclass
@dataclass(frozen=True)
class <DoSomething>Command:
aggregate_id: str # primitivos, validados en el adapter (Pydantic)
# ... otros campos
# NO: objetos de dominio como campos
# NO: métodos de negocio
# --- DTO de salida (Read Model) ---
# application/queries/<get_something>/read_model.py
@dataclass(frozen=True)
class <GetSomething>ReadModel:
id: str
status: str # valor primitivo, no el enum del dominio
# ... solo los campos que el consumidor necesita
# --- Mapper (Application Layer) ---
# application/queries/<get_something>/mapper.py (o inline en el handler)
from <project>.contexts.<bc>.domain.model.<aggregate> import <Aggregate>
from .read_model import <GetSomething>ReadModel
def to_read_model(aggregate: <Aggregate>) -> <GetSomething>ReadModel:
"""
Mapeo explícito: dominio → DTO de salida.
Cada campo mapeado es una decisión de contrato.
"""
return <GetSomething>ReadModel(
id=str(aggregate.id), # VO → string
status=aggregate.status.value, # Enum → string primitivo
# ... otros campos explícitos
)
# NO: return <GetSomething>ReadModel(**aggregate.__dict__) ← mapeo automático peligroso
# --- Mapper de entrada: Adapter HTTP → Command ---
# infrastructure/adapters/api/schemas/<do_something>_schema.py
from pydantic import BaseModel
class <DoSomething>RequestSchema(BaseModel):
aggregate_id: str
# ... validación Pydantic aquí
def to_command(self) -> <DoSomething>Command:
"""Adapter HTTP construye el Command. El dominio no sabe nada de Pydantic."""
return <DoSomething>Command(aggregate_id=self.aggregate_id)
# --- Mapeo en el Query Handler (para casos simples) ---
class <GetSomething>Handler:
def handle(self, query: <GetSomething>Query) -> list[<GetSomething>ReadModel]:
aggregates = self._repo.find_all()
return [to_read_model(a) for a in aggregates]Debugging lab
Detecta y corrige el error o la violación de diseño.
- 5.6.5.1
class GetOrderHandler: def handle(self, query): order = self._repo.get_by_id(OrderId(query.id)) return order # devuelve el aggregate directamente
- 5.6.5.2
def to_read_model(order: Order) -> OrderReadModel: return OrderReadModel(**order.__dict__) # mapeo automático
- 5.6.5.3
@dataclass(frozen=True) class CreateOrderCommand: customer: Customer # objeto de dominio como campo del Command
- 5.6.5.4
# En el aggregate: class Order: def to_dict(self) -> dict: return {'id': str(self.id), 'status': self.status.value}
- 5.6.5.5
@dataclass(frozen=True) class OrderReadModel: id: str status: OrderStatus # enum del dominio en el DTO de salida