← Volver al listado de tecnologías

Capítulo 5: Agregados

Por: Artiko
dddagregadospythontactical-design

Capítulo 5: Agregados

← Anterior Siguiente →

¿Qué es un Agregado?

Un Agregado es un cluster de objetos de dominio que se tratan como una unidad para propósitos de cambios de datos. Define un límite de consistencia transaccional.

Componentes del Agregado

graph TB
    subgraph Agregado["AGREGADO"]
        ROOT["🔷 AGGREGATE ROOT\n- ID global único\n- Punto de entrada único\n- Controla invariantes"]
        ENT["Entidad Interna\n(ID local)"]
        VO["Value Object\n(inmutable)"]

        ROOT -->|"contiene y protege"| ENT
        ROOT -->|"contiene y protege"| VO
    end

    EXT[/"Acceso Externo"/] -->|"solo a través de"| ROOT

“Un agregado es un cluster de objetos asociados que tratamos como una unidad para propósitos de cambios de datos.” — Eric Evans

Reglas Fundamentales

ReglaDescripción
IdentidadSolo la raíz tiene identidad global
AccesoTodo acceso externo pasa por la raíz
RepositorioSolo la raíz se obtiene del repositorio
ConsistenciaLa raíz garantiza invariantes del cluster
ReferenciasEntre agregados, solo por ID

Ejemplo Completo: Pedido

from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Optional
from uuid import UUID, uuid4

class EstadoPedido(Enum):
    BORRADOR = "borrador"
    CONFIRMADO = "confirmado"
    PAGADO = "pagado"
    EN_PREPARACION = "en_preparacion"
    ENVIADO = "enviado"
    ENTREGADO = "entregado"
    CANCELADO = "cancelado"

@dataclass
class LineaPedido:
    """
    Entidad interna del agregado Pedido.
    Su ID es local (solo único dentro del pedido).
    """
    id: int  # ID local, no UUID
    producto_id: UUID  # Referencia externa por ID
    nombre_producto: str  # Desnormalizado para independencia
    cantidad: int
    precio_unitario: Decimal

    def __post_init__(self):
        if self.cantidad <= 0:
            raise ValueError("Cantidad debe ser positiva")
        if self.precio_unitario < 0:
            raise ValueError("Precio no puede ser negativo")

    @property
    def subtotal(self) -> Decimal:
        return self.precio_unitario * self.cantidad

    def ajustar_cantidad(self, nueva_cantidad: int) -> None:
        if nueva_cantidad <= 0:
            raise ValueError("Cantidad debe ser positiva")
        self.cantidad = nueva_cantidad

@dataclass
class DireccionEnvio:
    """Value Object dentro del agregado"""
    calle: str
    ciudad: str
    codigo_postal: str
    pais: str

@dataclass
class Pedido:
    """
    Aggregate Root: punto de entrada único al agregado.
    Responsable de mantener la consistencia de todo el cluster.
    """
    id: UUID = field(default_factory=uuid4)
    cliente_id: UUID = None  # Referencia a otro agregado por ID
    _lineas: list[LineaPedido] = field(default_factory=list)
    _estado: EstadoPedido = EstadoPedido.BORRADOR
    direccion_envio: Optional[DireccionEnvio] = None
    fecha_creacion: datetime = field(default_factory=datetime.now)
    fecha_confirmacion: Optional[datetime] = None
    _siguiente_linea_id: int = 1
    _eventos: list = field(default_factory=list)

    # === Invariantes del Agregado ===
    MAX_LINEAS = 100
    MIN_MONTO_CONFIRMACION = Decimal("10.00")

    # === Operaciones que modifican el agregado ===

    def agregar_producto(
        self,
        producto_id: UUID,
        nombre: str,
        cantidad: int,
        precio: Decimal
    ) -> LineaPedido:
        """Agrega un producto al pedido"""
        self._validar_modificable()
        self._validar_limite_lineas()

        # Verificar si ya existe el producto
        linea_existente = self._buscar_linea_por_producto(producto_id)
        if linea_existente:
            linea_existente.ajustar_cantidad(
                linea_existente.cantidad + cantidad
            )
            return linea_existente

        linea = LineaPedido(
            id=self._siguiente_linea_id,
            producto_id=producto_id,
            nombre_producto=nombre,
            cantidad=cantidad,
            precio_unitario=precio,
        )
        self._lineas.append(linea)
        self._siguiente_linea_id += 1
        return linea

    def eliminar_linea(self, linea_id: int) -> None:
        """Elimina una línea del pedido"""
        self._validar_modificable()
        linea = self._buscar_linea(linea_id)
        if not linea:
            raise LineaNoEncontrada(linea_id)
        self._lineas.remove(linea)

    def actualizar_cantidad(self, linea_id: int, cantidad: int) -> None:
        """Actualiza cantidad de una línea"""
        self._validar_modificable()
        linea = self._buscar_linea(linea_id)
        if not linea:
            raise LineaNoEncontrada(linea_id)
        linea.ajustar_cantidad(cantidad)

    def asignar_direccion(self, direccion: DireccionEnvio) -> None:
        """Asigna dirección de envío"""
        self._validar_modificable()
        self.direccion_envio = direccion

    # === Transiciones de estado ===

    def confirmar(self) -> None:
        """Confirma el pedido si cumple todas las reglas"""
        self._validar_puede_confirmarse()
        self._estado = EstadoPedido.CONFIRMADO
        self.fecha_confirmacion = datetime.now()
        self._eventos.append(PedidoConfirmado(self.id, self.total))

    def marcar_pagado(self) -> None:
        if self._estado != EstadoPedido.CONFIRMADO:
            raise TransicionInvalida("Solo pedidos confirmados pueden pagarse")
        self._estado = EstadoPedido.PAGADO
        self._eventos.append(PedidoPagado(self.id))

    def iniciar_preparacion(self) -> None:
        if self._estado != EstadoPedido.PAGADO:
            raise TransicionInvalida("Solo pedidos pagados pueden prepararse")
        self._estado = EstadoPedido.EN_PREPARACION

    def cancelar(self, motivo: str) -> None:
        """Cancela el pedido si es posible"""
        if self._estado in (EstadoPedido.ENVIADO, EstadoPedido.ENTREGADO):
            raise PedidoNoCancelable("Pedido ya fue enviado")
        self._estado = EstadoPedido.CANCELADO
        self._eventos.append(PedidoCancelado(self.id, motivo))

    # === Queries (no modifican estado) ===

    @property
    def estado(self) -> EstadoPedido:
        return self._estado

    @property
    def total(self) -> Decimal:
        return sum((l.subtotal for l in self._lineas), Decimal("0"))

    @property
    def cantidad_items(self) -> int:
        return sum(l.cantidad for l in self._lineas)

    @property
    def cantidad_lineas(self) -> int:
        return len(self._lineas)

    def obtener_lineas(self) -> list[LineaPedido]:
        """Retorna copia de las líneas para evitar modificación externa"""
        return list(self._lineas)

    def obtener_eventos(self) -> list:
        eventos = list(self._eventos)
        self._eventos.clear()
        return eventos

    # === Validaciones privadas ===

    def _validar_modificable(self) -> None:
        if self._estado != EstadoPedido.BORRADOR:
            raise PedidoNoModificable(
                f"Pedido en estado {self._estado.value} no es modificable"
            )

    def _validar_limite_lineas(self) -> None:
        if len(self._lineas) >= self.MAX_LINEAS:
            raise LimiteLineasExcedido(self.MAX_LINEAS)

    def _validar_puede_confirmarse(self) -> None:
        errores = []
        if self._estado != EstadoPedido.BORRADOR:
            errores.append("Pedido no está en borrador")
        if not self._lineas:
            errores.append("Pedido sin productos")
        if self.total < self.MIN_MONTO_CONFIRMACION:
            errores.append(f"Monto mínimo: {self.MIN_MONTO_CONFIRMACION}")
        if not self.direccion_envio:
            errores.append("Falta dirección de envío")
        if errores:
            raise PedidoNoConfirmable(errores)

    def _buscar_linea(self, linea_id: int) -> Optional[LineaPedido]:
        for linea in self._lineas:
            if linea.id == linea_id:
                return linea
        return None

    def _buscar_linea_por_producto(self, producto_id: UUID) -> Optional[LineaPedido]:
        for linea in self._lineas:
            if linea.producto_id == producto_id:
                return linea
        return None

Límites del Agregado

┌─────────────────────────────────────────────────────────────────┐
│                    LÍMITE DE TRANSACCIÓN                        │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  Agregado: Pedido                                               │
│  ┌────────────────────────────────────────────────────────┐    │
│  │  Pedido (Root)                                          │    │
│  │    │                                                    │    │
│  │    ├── LineaPedido (entidad interna)                   │    │
│  │    ├── LineaPedido (entidad interna)                   │    │
│  │    └── DireccionEnvio (value object)                   │    │
│  └────────────────────────────────────────────────────────┘    │
│                         │                                        │
│                         │ solo IDs                               │
│                         ▼                                        │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │  cliente_id  │  │ producto_id  │  │ producto_id  │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
│         │                 │                 │                    │
└─────────┼─────────────────┼─────────────────┼────────────────────┘
          │                 │                 │
          ▼                 ▼                 ▼
    ┌──────────┐      ┌──────────┐      ┌──────────┐
    │ Cliente  │      │ Producto │      │ Producto │
    │ (otro    │      │ (otro    │      │ (otro    │
    │ agregado)│      │ agregado)│      │ agregado)│
    └──────────┘      └──────────┘      └──────────┘

Consistencia Eventual entre Agregados

# Cuando necesitas actualizar múltiples agregados, usa eventos

@dataclass
class PedidoConfirmado:
    pedido_id: UUID
    total: Decimal
    timestamp: datetime = field(default_factory=datetime.now)

@dataclass
class PedidoCancelado:
    pedido_id: UUID
    motivo: str
    timestamp: datetime = field(default_factory=datetime.now)

class ManejadorEventosPedido:
    """Procesa eventos para mantener consistencia eventual"""

    def __init__(
        self,
        repo_inventario: "RepositorioInventario",
        notificador: "Notificador"
    ):
        self.repo_inventario = repo_inventario
        self.notificador = notificador

    def al_confirmar_pedido(self, evento: PedidoConfirmado) -> None:
        """Reserva inventario cuando se confirma un pedido"""
        # Esto ocurre en otra transacción
        pedido = self.repo_pedidos.obtener(evento.pedido_id)
        for linea in pedido.obtener_lineas():
            self.repo_inventario.reservar(
                linea.producto_id, linea.cantidad
            )

    def al_cancelar_pedido(self, evento: PedidoCancelado) -> None:
        """Libera inventario cuando se cancela un pedido"""
        # Otra transacción
        pedido = self.repo_pedidos.obtener(evento.pedido_id)
        for linea in pedido.obtener_lineas():
            self.repo_inventario.liberar(
                linea.producto_id, linea.cantidad
            )

Ejemplo: Agregado Reserva (Hotel)

from dataclasses import dataclass, field
from datetime import date, timedelta
from decimal import Decimal
from enum import Enum
from uuid import UUID, uuid4

class EstadoReserva(Enum):
    PENDIENTE = "pendiente"
    CONFIRMADA = "confirmada"
    CHECKIN = "checkin"
    CHECKOUT = "checkout"
    CANCELADA = "cancelada"
    NO_SHOW = "no_show"

@dataclass(frozen=True)
class PeriodoEstancia:
    """Value Object: período de la reserva"""
    entrada: date
    salida: date

    def __post_init__(self):
        if self.entrada >= self.salida:
            raise ValueError("Entrada debe ser antes de salida")

    @property
    def noches(self) -> int:
        return (self.salida - self.entrada).days

    def se_superpone(self, otro: "PeriodoEstancia") -> bool:
        return self.entrada < otro.salida and otro.entrada < self.salida

@dataclass
class ServicioAdicional:
    """Entidad interna: servicios contratados"""
    id: int
    tipo: str  # "desayuno", "spa", "parking"
    fecha: date
    precio: Decimal
    cantidad: int = 1

    @property
    def total(self) -> Decimal:
        return self.precio * self.cantidad

@dataclass
class Reserva:
    """Aggregate Root para reservas de hotel"""
    id: UUID = field(default_factory=uuid4)
    huesped_id: UUID = None
    habitacion_id: UUID = None  # Referencia por ID
    periodo: PeriodoEstancia = None
    _servicios: list[ServicioAdicional] = field(default_factory=list)
    _estado: EstadoReserva = EstadoReserva.PENDIENTE
    tarifa_noche: Decimal = Decimal("0")
    _siguiente_servicio_id: int = 1

    def confirmar(self) -> None:
        if self._estado != EstadoReserva.PENDIENTE:
            raise TransicionInvalida("Solo reservas pendientes")
        self._estado = EstadoReserva.CONFIRMADA

    def hacer_checkin(self, fecha: date) -> None:
        if self._estado != EstadoReserva.CONFIRMADA:
            raise TransicionInvalida("Reserva no confirmada")
        if fecha < self.periodo.entrada:
            raise CheckinTemprano()
        self._estado = EstadoReserva.CHECKIN

    def hacer_checkout(self) -> Decimal:
        if self._estado != EstadoReserva.CHECKIN:
            raise TransicionInvalida("No hay checkin activo")
        self._estado = EstadoReserva.CHECKOUT
        return self.total

    def agregar_servicio(
        self, tipo: str, fecha: date, precio: Decimal, cantidad: int = 1
    ) -> None:
        if self._estado != EstadoReserva.CHECKIN:
            raise ReservaNoActiva()
        if fecha < self.periodo.entrada or fecha >= self.periodo.salida:
            raise FechaFueraDeEstancia()

        servicio = ServicioAdicional(
            id=self._siguiente_servicio_id,
            tipo=tipo,
            fecha=fecha,
            precio=precio,
            cantidad=cantidad
        )
        self._servicios.append(servicio)
        self._siguiente_servicio_id += 1

    @property
    def total_alojamiento(self) -> Decimal:
        return self.tarifa_noche * self.periodo.noches

    @property
    def total_servicios(self) -> Decimal:
        return sum((s.total for s in self._servicios), Decimal("0"))

    @property
    def total(self) -> Decimal:
        return self.total_alojamiento + self.total_servicios

Tamaño del Agregado

┌─────────────────────────────────────────────────────────────┐
│                 TAMAÑO DEL AGREGADO                         │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  PEQUEÑO (recomendado)           GRANDE (evitar)            │
│  ┌──────────────┐                ┌──────────────────────┐   │
│  │   Root       │                │   Root               │   │
│  │   ├── VO     │                │   ├── Entidad        │   │
│  │   └── VO     │                │   │   ├── Entidad    │   │
│  └──────────────┘                │   │   └── Entidad    │   │
│                                  │   ├── Entidad        │   │
│  ✓ Menos conflictos             │   │   └── VO         │   │
│  ✓ Transacciones rápidas        │   └── Entidad        │   │
│  ✓ Fácil de entender            │       └── Entidad    │   │
│                                  └──────────────────────┘   │
│                                                              │
│                                  ✗ Conflictos frecuentes    │
│                                  ✗ Transacciones largas     │
│                                  ✗ Difícil de mantener      │
└─────────────────────────────────────────────────────────────┘

Antipatrones a Evitar

1. Agregado Anémico

# MAL: Agregado sin lógica de negocio
@dataclass
class PedidoAnemico:
    id: UUID
    lineas: list[LineaPedido]  # Expone lista directamente
    estado: str  # String en lugar de enum

# El servicio hace todo el trabajo
class PedidoService:
    def agregar_linea(self, pedido, producto, cantidad):
        pedido.lineas.append(...)  # Violación de encapsulación
        pedido.estado = "confirmado"  # Sin validación

# BIEN: Lógica en el agregado

2. Referencias Directas entre Agregados

# MAL: Referencia directa
@dataclass
class Pedido:
    cliente: Cliente  # Objeto completo

# BIEN: Solo ID
@dataclass
class Pedido:
    cliente_id: UUID  # Solo el identificador

3. Transacciones que Cruzan Agregados

# MAL: Una transacción modifica múltiples agregados
def confirmar_pedido(pedido_id: UUID):
    with transaccion():
        pedido = repo_pedidos.obtener(pedido_id)
        cliente = repo_clientes.obtener(pedido.cliente_id)
        inventario = repo_inventario.obtener_todos()

        pedido.confirmar()
        cliente.agregar_pedido(pedido)  # Otro agregado
        for linea in pedido.lineas:
            inventario.reservar(linea.producto_id)  # Otro agregado

        repo_pedidos.guardar(pedido)
        repo_clientes.guardar(cliente)
        repo_inventario.guardar_todos(inventario)

# BIEN: Una transacción por agregado + eventos
def confirmar_pedido(pedido_id: UUID):
    with transaccion():
        pedido = repo_pedidos.obtener(pedido_id)
        pedido.confirmar()
        repo_pedidos.guardar(pedido)
        # Emite evento PedidoConfirmado

    # Los manejadores de eventos actualizan otros agregados

Checklist de Agregados

← Anterior Siguiente →
← Volver al listado de tecnologías