← Volver al listado de tecnologías
Capítulo 2: Bounded Contexts
Capítulo 2: Bounded Contexts
¿Qué es un Bounded Context?
Un Bounded Context es una frontera semántica donde un modelo de dominio es válido y consistente. Dentro de esta frontera:
- El lenguaje ubicuo tiene significado preciso
- Las reglas de negocio son coherentes
- El modelo representa la realidad del subdominio
“Divide explícitamente modelos grandes. Sé explícito sobre el contexto en el que aplica un modelo.” — Eric Evans
El Problema: Modelos Unificados
En un e-commerce, “Producto” significa cosas diferentes según el contexto:
classDiagram
class ProductoUnificado {
+nombre
+descripcion
+imagenes
+categoria
+sku
+cantidad
+ubicacion_almacen
+precio
+descuento
+peso
+dimensiones
+costo
+depreciacion
}
note for ProductoUnificado "❌ MODELO UNIFICADO (MAL)\nClase gigante, acoplada,\nimposible de mantener"Solución: Separar Contextos
graph TB
subgraph Catalogo["CATÁLOGO"]
P1[Producto]
P1 --> |nombre| A1[ ]
P1 --> |descripcion| A2[ ]
P1 --> |imagenes| A3[ ]
end
subgraph Inventario["INVENTARIO"]
P2[ArticuloStock]
P2 --> |sku| B1[ ]
P2 --> |cantidad| B2[ ]
P2 --> |ubicacion| B3[ ]
end
subgraph Ventas["VENTAS"]
P3[ProductoVenta]
P3 --> |precio| C1[ ]
P3 --> |descuento| C2[ ]
P3 --> |impuestos| C3[ ]
end
P1 -.->|producto_id| REF((ID))
P2 -.->|producto_id| REF
P3 -.->|producto_id| REFImplementación por Contexto
Contexto: Catálogo
# catalogo/domain/entities.py
from dataclasses import dataclass, field
from typing import Optional
from uuid import UUID, uuid4
@dataclass
class Categoria:
id: UUID
nombre: str
padre: Optional["Categoria"] = None
def es_subcategoria_de(self, categoria: "Categoria") -> bool:
actual = self.padre
while actual:
if actual.id == categoria.id:
return True
actual = actual.padre
return False
@dataclass
class Producto:
id: UUID = field(default_factory=uuid4)
nombre: str = ""
descripcion: str = ""
categoria: Categoria = None
imagenes: list[str] = field(default_factory=list)
atributos: dict[str, str] = field(default_factory=dict)
esta_publicado: bool = False
def publicar(self) -> None:
if not self._es_valido_para_publicar():
raise ProductoInvalidoError("Faltan datos requeridos")
self.esta_publicado = True
def _es_valido_para_publicar(self) -> bool:
return (
len(self.nombre) >= 3
and len(self.descripcion) >= 10
and self.categoria is not None
and len(self.imagenes) >= 1
)
def agregar_imagen(self, url: str) -> None:
if len(self.imagenes) >= 10:
raise LimiteImagenesError("Máximo 10 imágenes")
self.imagenes.append(url)
def actualizar_atributo(self, clave: str, valor: str) -> None:
self.atributos[clave] = valorContexto: Inventario
# inventario/domain/entities.py
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from enum import Enum
from uuid import UUID, uuid4
class TipoMovimiento(Enum):
ENTRADA = "entrada"
SALIDA = "salida"
AJUSTE = "ajuste"
RESERVA = "reserva"
LIBERACION = "liberacion"
@dataclass
class Ubicacion:
almacen: str
pasillo: str
estante: str
nivel: int
def codigo(self) -> str:
return f"{self.almacen}-{self.pasillo}-{self.estante}-{self.nivel}"
@dataclass
class MovimientoStock:
id: UUID = field(default_factory=uuid4)
tipo: TipoMovimiento = None
cantidad: int = 0
fecha: datetime = field(default_factory=datetime.now)
referencia: str = "" # pedido_id, ajuste_id, etc.
usuario: str = ""
@dataclass
class ArticuloStock:
sku: str
producto_id: UUID # Referencia al contexto Catálogo
cantidad_disponible: int = 0
cantidad_reservada: int = 0
ubicacion: Ubicacion = None
punto_reorden: int = 10
movimientos: list[MovimientoStock] = field(default_factory=list)
@property
def cantidad_total(self) -> int:
return self.cantidad_disponible + self.cantidad_reservada
def reservar(self, cantidad: int, pedido_id: str) -> None:
"""Reserva stock para un pedido"""
if cantidad > self.cantidad_disponible:
raise StockInsuficienteError(
f"Disponible: {self.cantidad_disponible}, Solicitado: {cantidad}"
)
self.cantidad_disponible -= cantidad
self.cantidad_reservada += cantidad
self._registrar_movimiento(TipoMovimiento.RESERVA, cantidad, pedido_id)
def liberar_reserva(self, cantidad: int, pedido_id: str) -> None:
"""Libera stock reservado (cancelación de pedido)"""
self.cantidad_reservada -= cantidad
self.cantidad_disponible += cantidad
self._registrar_movimiento(TipoMovimiento.LIBERACION, cantidad, pedido_id)
def confirmar_salida(self, cantidad: int, pedido_id: str) -> None:
"""Confirma la salida del stock reservado"""
self.cantidad_reservada -= cantidad
self._registrar_movimiento(TipoMovimiento.SALIDA, cantidad, pedido_id)
def recibir_stock(self, cantidad: int, referencia: str) -> None:
"""Registra entrada de nuevo stock"""
self.cantidad_disponible += cantidad
self._registrar_movimiento(TipoMovimiento.ENTRADA, cantidad, referencia)
def necesita_reorden(self) -> bool:
return self.cantidad_total <= self.punto_reorden
def _registrar_movimiento(
self, tipo: TipoMovimiento, cantidad: int, referencia: str
) -> None:
self.movimientos.append(
MovimientoStock(tipo=tipo, cantidad=cantidad, referencia=referencia)
)Contexto: Ventas
# ventas/domain/entities.py
from dataclasses import dataclass, field
from decimal import Decimal
from enum import Enum
from uuid import UUID, uuid4
class TipoDescuento(Enum):
PORCENTAJE = "porcentaje"
MONTO_FIJO = "monto_fijo"
@dataclass
class Descuento:
tipo: TipoDescuento
valor: Decimal
codigo: str = ""
def aplicar(self, precio: Decimal) -> Decimal:
if self.tipo == TipoDescuento.PORCENTAJE:
return precio * (self.valor / Decimal("100"))
return min(self.valor, precio)
@dataclass
class ProductoVenta:
"""Producto en el contexto de ventas - solo lo relevante para vender"""
id: UUID
producto_id: UUID # Referencia a Catálogo
nombre: str # Desnormalizado para rendimiento
precio_base: Decimal
precio_actual: Decimal
descuento: Descuento = None
impuesto_porcentaje: Decimal = Decimal("16")
@property
def precio_con_descuento(self) -> Decimal:
if self.descuento:
return self.precio_actual - self.descuento.aplicar(self.precio_actual)
return self.precio_actual
@property
def precio_final(self) -> Decimal:
"""Precio con descuento e impuestos incluidos"""
precio = self.precio_con_descuento
impuesto = precio * (self.impuesto_porcentaje / Decimal("100"))
return precio + impuesto
def tiene_descuento_activo(self) -> bool:
return self.descuento is not None and self.descuento.valor > 0Context Mapping: Relaciones entre Contextos
┌─────────────────────────────────────────────────────────────┐
│ CONTEXT MAP │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ Shared Kernel ┌──────────┐ │
│ │ Catálogo │◄────────────────────────│ Ventas │ │
│ └────┬─────┘ (producto_id) └────┬─────┘ │
│ │ │ │
│ │ Customer-Supplier │ │
│ │ (Catálogo provee) │ │
│ ▼ │ │
│ ┌──────────┐ │ │
│ │Inventario│◄─────────────────────────────┘ │
│ └────┬─────┘ Anticorruption Layer │
│ │ │
│ │ Conformist │
│ ▼ │
│ ┌──────────┐ │
│ │ Envíos │ │
│ └──────────┘ │
└─────────────────────────────────────────────────────────────┘Tipos de Relaciones
| Relación | Descripción | Uso |
|---|---|---|
| Shared Kernel | Código compartido entre contextos | IDs, tipos básicos |
| Customer-Supplier | Un contexto depende de otro | Inventario consume Catálogo |
| Conformist | Adopta modelo del upstream | Envíos usa modelo de Inventario |
| Anticorruption Layer | Traduce entre modelos | Ventas traduce de Catálogo |
| Open Host Service | API pública del contexto | API REST de Catálogo |
| Published Language | Formato común de intercambio | JSON schemas, eventos |
Anticorruption Layer (ACL)
El ACL protege tu dominio de modelos externos:
# ventas/infrastructure/adapters/catalogo_adapter.py
from dataclasses import dataclass
from decimal import Decimal
from uuid import UUID
# Cliente externo (podría ser HTTP, gRPC, etc.)
from ventas.infrastructure.clients import CatalogoClient
from ventas.domain.entities import ProductoVenta
@dataclass
class CatalogoAdapter:
"""
Anticorruption Layer: Traduce el modelo de Catálogo
al modelo de Ventas, protegiendo el dominio de Ventas
de cambios en Catálogo.
"""
cliente: CatalogoClient
servicio_precios: "ServicioPrecios"
def obtener_producto_para_venta(self, producto_id: UUID) -> ProductoVenta:
# Obtener datos crudos del contexto externo
datos_externos = self.cliente.obtener_producto(str(producto_id))
# Traducir al modelo de nuestro dominio
return self._traducir_a_producto_venta(datos_externos)
def _traducir_a_producto_venta(self, datos: dict) -> ProductoVenta:
"""Traduce el modelo externo a nuestro modelo interno"""
precio = self.servicio_precios.obtener_precio_actual(datos["id"])
return ProductoVenta(
id=UUID(datos["id"]),
producto_id=UUID(datos["id"]),
nombre=datos["nombre"],
precio_base=Decimal(str(datos.get("precio_base", 0))),
precio_actual=precio,
descuento=self._traducir_descuento(datos.get("promocion")),
)
def _traducir_descuento(self, promocion: dict | None) -> Descuento | None:
if not promocion:
return None
return Descuento(
tipo=TipoDescuento(promocion["tipo"]),
valor=Decimal(str(promocion["valor"])),
codigo=promocion.get("codigo", ""),
)Comunicación entre Contextos
Eventos de Dominio
# shared/events.py
from dataclasses import dataclass, field
from datetime import datetime
from uuid import UUID, uuid4
@dataclass
class EventoDominio:
id: UUID = field(default_factory=uuid4)
ocurrido_en: datetime = field(default_factory=datetime.now)
version: int = 1
@dataclass
class ProductoPublicado(EventoDominio):
"""Emitido por Catálogo cuando un producto se publica"""
producto_id: UUID = None
nombre: str = ""
categoria_id: UUID = None
@dataclass
class StockAgotado(EventoDominio):
"""Emitido por Inventario cuando el stock llega a cero"""
sku: str = ""
producto_id: UUID = None
almacen: str = ""
# ventas/application/handlers.py
class ManejadorEventosCatalogo:
def __init__(self, repo_productos: RepositorioProductosVenta):
self.repo = repo_productos
def al_publicar_producto(self, evento: ProductoPublicado) -> None:
"""Crea ProductoVenta cuando Catálogo publica un producto"""
producto_venta = ProductoVenta(
id=uuid4(),
producto_id=evento.producto_id,
nombre=evento.nombre,
precio_base=Decimal("0"), # Se configura después
precio_actual=Decimal("0"),
)
self.repo.guardar(producto_venta)Estructura de Carpetas
proyecto/
├── shared/ # Shared Kernel
│ ├── domain/
│ │ ├── value_objects.py # Money, Email, etc.
│ │ └── events.py # Eventos base
│ └── infrastructure/
│ └── messaging.py # Bus de eventos
│
├── catalogo/ # Bounded Context
│ ├── domain/
│ │ ├── entities.py
│ │ ├── repositories.py # Interfaces
│ │ └── services.py
│ ├── application/
│ │ ├── commands.py
│ │ ├── queries.py
│ │ └── handlers.py
│ └── infrastructure/
│ ├── persistence/
│ └── api/
│
├── inventario/ # Bounded Context
│ └── ... (misma estructura)
│
└── ventas/ # Bounded Context
└── ... (misma estructura)Checklist de Bounded Context
- ¿El contexto tiene un propósito claro y acotado?
- ¿El lenguaje ubicuo es consistente dentro del contexto?
- ¿Las referencias a otros contextos son solo por ID?
- ¿Existe un ACL para traducir modelos externos?
- ¿Los eventos de dominio comunican cambios importantes?