← Volver al listado de tecnologías

Capítulo 7: Servicios de Dominio

Por: Artiko
dddserviciospythontactical-design

Capítulo 7: Servicios de Dominio

¿Qué es un Servicio de Dominio?

Un Servicio de Dominio encapsula lógica de negocio que no pertenece naturalmente a ninguna entidad o value object. Es una operación significativa del dominio que:

┌─────────────────────────────────────────────────────────────┐
│           CUÁNDO USAR SERVICIO DE DOMINIO                   │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ¿La operación pertenece claramente a una entidad?          │
│                     │                                        │
│         ┌───────────┴───────────┐                           │
│         ▼                       ▼                           │
│        SÍ                      NO                           │
│         │                       │                           │
│         ▼                       ▼                           │
│  Método en la            ¿Involucra múltiples              │
│  entidad                  agregados?                        │
│                                 │                           │
│                    ┌────────────┴────────────┐              │
│                    ▼                         ▼              │
│                   SÍ                        NO              │
│                    │                         │              │
│                    ▼                         ▼              │
│             SERVICIO DE              Value Object           │
│             DOMINIO                  o función pura         │
│                                                              │
└─────────────────────────────────────────────────────────────┘

“Cuando una operación significativa del dominio no pertenece conceptualmente a ningún objeto, es apropiado ponerla en un servicio.” — Eric Evans

Características de Servicios de Dominio

CaracterísticaDescripción
Sin estadoNo almacenan datos entre llamadas
Sin identidadNo tienen ID, son intercambiables
Operación puraReciben todo lo que necesitan como parámetros
Lenguaje ubicuoSu nombre refleja un concepto del dominio
Sin infraestructuraNo acceden a BD, no envían emails

Ejemplo: Transferencia entre Cuentas

from dataclasses import dataclass
from decimal import Decimal
from datetime import datetime

@dataclass
class ResultadoTransferencia:
    exitosa: bool
    referencia: str
    fecha: datetime
    mensaje: str = ""

class ServicioTransferencias:
    """
    Servicio de Dominio para transferencias bancarias.
    La transferencia involucra dos cuentas - no pertenece a ninguna.
    """

    def transferir(
        self,
        cuenta_origen: "Cuenta",
        cuenta_destino: "Cuenta",
        monto: "Dinero",
    ) -> ResultadoTransferencia:
        # Validaciones de negocio
        errores = self._validar_transferencia(
            cuenta_origen, cuenta_destino, monto
        )
        if errores:
            return ResultadoTransferencia(
                exitosa=False,
                referencia="",
                fecha=datetime.now(),
                mensaje="; ".join(errores)
            )

        # Ejecutar transferencia
        cuenta_origen.debitar(monto)
        cuenta_destino.acreditar(monto)

        return ResultadoTransferencia(
            exitosa=True,
            referencia=self._generar_referencia(),
            fecha=datetime.now(),
        )

    def _validar_transferencia(
        self,
        origen: "Cuenta",
        destino: "Cuenta",
        monto: "Dinero"
    ) -> list[str]:
        errores = []

        if origen.id == destino.id:
            errores.append("No puede transferir a la misma cuenta")

        if not origen.esta_activa():
            errores.append("Cuenta origen no está activa")

        if not destino.esta_activa():
            errores.append("Cuenta destino no está activa")

        if not origen.puede_debitar(monto):
            errores.append("Saldo insuficiente en cuenta origen")

        if monto.cantidad <= 0:
            errores.append("Monto debe ser positivo")

        limite_diario = origen.limite_transferencia_diaria
        if monto.cantidad > limite_diario:
            errores.append(f"Excede límite diario de {limite_diario}")

        return errores

    def _generar_referencia(self) -> str:
        from uuid import uuid4
        return f"TRF-{uuid4().hex[:8].upper()}"

Ejemplo: Cálculo de Precios

from decimal import Decimal
from dataclasses import dataclass
from enum import Enum

class TipoCliente(Enum):
    REGULAR = "regular"
    PREMIUM = "premium"
    VIP = "vip"
    CORPORATIVO = "corporativo"

@dataclass
class DetallesPrecio:
    """Value Object con el desglose del precio"""
    subtotal: Decimal
    descuento_cliente: Decimal
    descuento_volumen: Decimal
    descuento_promocional: Decimal
    impuestos: Decimal
    total: Decimal

    @property
    def descuento_total(self) -> Decimal:
        return (
            self.descuento_cliente +
            self.descuento_volumen +
            self.descuento_promocional
        )

class ServicioPrecios:
    """Servicio de Dominio para cálculo de precios"""

    TASAS_CLIENTE = {
        TipoCliente.REGULAR: Decimal("0"),
        TipoCliente.PREMIUM: Decimal("0.10"),
        TipoCliente.VIP: Decimal("0.20"),
        TipoCliente.CORPORATIVO: Decimal("0.15"),
    }

    UMBRALES_VOLUMEN = [
        (100, Decimal("0.15")),
        (50, Decimal("0.10")),
        (20, Decimal("0.05")),
    ]

    TASA_IMPUESTO = Decimal("0.16")

    def calcular_precio(
        self,
        producto: "Producto",
        cliente: "Cliente",
        cantidad: int,
        promocion: "Promocion | None" = None,
    ) -> DetallesPrecio:
        subtotal = producto.precio * cantidad

        descuento_cliente = self._calcular_descuento_cliente(
            subtotal, cliente.tipo
        )
        descuento_volumen = self._calcular_descuento_volumen(
            subtotal, cantidad
        )
        descuento_promocional = self._calcular_descuento_promocion(
            subtotal, promocion
        )

        # Solo aplicar el mayor descuento
        mejor_descuento = max(
            descuento_cliente, descuento_volumen, descuento_promocional
        )
        precio_con_descuento = subtotal - mejor_descuento
        impuestos = precio_con_descuento * self.TASA_IMPUESTO

        return DetallesPrecio(
            subtotal=subtotal,
            descuento_cliente=descuento_cliente if descuento_cliente == mejor_descuento else Decimal("0"),
            descuento_volumen=descuento_volumen if descuento_volumen == mejor_descuento else Decimal("0"),
            descuento_promocional=descuento_promocional if descuento_promocional == mejor_descuento else Decimal("0"),
            impuestos=impuestos,
            total=precio_con_descuento + impuestos,
        )

    def _calcular_descuento_cliente(
        self, monto: Decimal, tipo: TipoCliente
    ) -> Decimal:
        tasa = self.TASAS_CLIENTE.get(tipo, Decimal("0"))
        return monto * tasa

    def _calcular_descuento_volumen(
        self, monto: Decimal, cantidad: int
    ) -> Decimal:
        for umbral, tasa in self.UMBRALES_VOLUMEN:
            if cantidad >= umbral:
                return monto * tasa
        return Decimal("0")

    def _calcular_descuento_promocion(
        self, monto: Decimal, promocion: "Promocion | None"
    ) -> Decimal:
        if not promocion or not promocion.esta_vigente():
            return Decimal("0")
        return promocion.calcular_descuento(monto)

Ejemplo: Validación de Disponibilidad

from dataclasses import dataclass
from uuid import UUID

@dataclass
class ResultadoDisponibilidad:
    disponible: bool
    cantidad_disponible: int
    almacenes_con_stock: list["Almacen"]
    cantidad_faltante: int = 0
    alternativas: list["Producto"] = None

    def __post_init__(self):
        if self.alternativas is None:
            self.alternativas = []

class ServicioDisponibilidad:
    """Verifica disponibilidad de productos en múltiples almacenes"""

    def verificar_disponibilidad(
        self,
        producto: "Producto",
        cantidad: int,
        almacenes: list["Almacen"],
    ) -> ResultadoDisponibilidad:
        stock_por_almacen = {
            a: a.stock_de(producto.id) for a in almacenes
        }
        stock_total = sum(stock_por_almacen.values())
        almacenes_con_stock = [
            a for a, s in stock_por_almacen.items() if s > 0
        ]

        if stock_total >= cantidad:
            return ResultadoDisponibilidad(
                disponible=True,
                cantidad_disponible=stock_total,
                almacenes_con_stock=almacenes_con_stock,
            )

        return ResultadoDisponibilidad(
            disponible=False,
            cantidad_disponible=stock_total,
            almacenes_con_stock=almacenes_con_stock,
            cantidad_faltante=cantidad - stock_total,
        )

    def verificar_disponibilidad_pedido(
        self,
        pedido: "Pedido",
        almacenes: list["Almacen"],
    ) -> dict[UUID, ResultadoDisponibilidad]:
        """Verifica disponibilidad de todos los productos del pedido"""
        resultados = {}
        for linea in pedido.obtener_lineas():
            producto = self._obtener_producto(linea.producto_id)
            resultados[linea.producto_id] = self.verificar_disponibilidad(
                producto, linea.cantidad, almacenes
            )
        return resultados

    def pedido_completamente_disponible(
        self, resultados: dict[UUID, ResultadoDisponibilidad]
    ) -> bool:
        return all(r.disponible for r in resultados.values())

Ejemplo: Cálculo de Envíos

from dataclasses import dataclass
from decimal import Decimal
from enum import Enum

class MetodoEnvio(Enum):
    ESTANDAR = "estandar"
    EXPRESS = "express"
    MISMO_DIA = "mismo_dia"

@dataclass(frozen=True)
class TarifaEnvio:
    metodo: MetodoEnvio
    base: Decimal
    por_kg: Decimal
    por_km: Decimal
    dias_estimados: int

@dataclass
class CostoEnvio:
    metodo: MetodoEnvio
    costo_base: Decimal
    costo_peso: Decimal
    costo_distancia: Decimal
    total: Decimal
    dias_estimados: int
    gratis: bool = False

class ServicioEnvios:
    """Calcula costos de envío basado en peso, distancia y método"""

    TARIFAS = {
        MetodoEnvio.ESTANDAR: TarifaEnvio(
            MetodoEnvio.ESTANDAR,
            base=Decimal("50"),
            por_kg=Decimal("10"),
            por_km=Decimal("0.5"),
            dias_estimados=5
        ),
        MetodoEnvio.EXPRESS: TarifaEnvio(
            MetodoEnvio.EXPRESS,
            base=Decimal("100"),
            por_kg=Decimal("15"),
            por_km=Decimal("1"),
            dias_estimados=2
        ),
        MetodoEnvio.MISMO_DIA: TarifaEnvio(
            MetodoEnvio.MISMO_DIA,
            base=Decimal("200"),
            por_kg=Decimal("25"),
            por_km=Decimal("2"),
            dias_estimados=0
        ),
    }

    MONTO_ENVIO_GRATIS = Decimal("1000")

    def calcular_costo(
        self,
        pedido: "Pedido",
        direccion_destino: "Direccion",
        direccion_origen: "Direccion",
        metodo: MetodoEnvio,
    ) -> CostoEnvio:
        tarifa = self.TARIFAS[metodo]
        peso_total = self._calcular_peso_pedido(pedido)
        distancia = self._calcular_distancia(direccion_origen, direccion_destino)

        costo_peso = peso_total * tarifa.por_kg
        costo_distancia = Decimal(str(distancia)) * tarifa.por_km
        total = tarifa.base + costo_peso + costo_distancia

        # Envío gratis para pedidos grandes
        envio_gratis = pedido.total >= self.MONTO_ENVIO_GRATIS

        return CostoEnvio(
            metodo=metodo,
            costo_base=tarifa.base,
            costo_peso=costo_peso,
            costo_distancia=costo_distancia,
            total=Decimal("0") if envio_gratis else total,
            dias_estimados=tarifa.dias_estimados,
            gratis=envio_gratis,
        )

    def _calcular_peso_pedido(self, pedido: "Pedido") -> Decimal:
        return sum(
            linea.cantidad * Decimal("0.5")  # Peso promedio por item
            for linea in pedido.obtener_lineas()
        )

    def _calcular_distancia(
        self, origen: "Direccion", destino: "Direccion"
    ) -> float:
        # Cálculo simplificado
        if origen.ciudad == destino.ciudad:
            return 10
        if origen.estado == destino.estado:
            return 100
        return 500

    def obtener_opciones_envio(
        self, pedido: "Pedido", origen: "Direccion", destino: "Direccion"
    ) -> list[CostoEnvio]:
        """Retorna todas las opciones de envío disponibles"""
        return [
            self.calcular_costo(pedido, destino, origen, metodo)
            for metodo in MetodoEnvio
        ]

Servicio de Dominio vs Caso de Uso

┌─────────────────────────────────────────────────────────────┐
│          SERVICIO DE DOMINIO vs CASO DE USO                 │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  SERVICIO DE DOMINIO              CASO DE USO               │
│  (Capa Dominio)                   (Capa Aplicación)         │
│                                                              │
│  ┌──────────────────┐            ┌──────────────────┐       │
│  │ ServicioPrecios  │            │ ConfirmarPedido  │       │
│  │                  │            │                  │       │
│  │ - calcular_      │            │ - repositorios   │       │
│  │   precio()       │            │ - servicios      │       │
│  │                  │            │ - eventos        │       │
│  │ Sin estado       │            │ Orquesta flujo   │       │
│  │ Sin I/O          │            │ Usa I/O          │       │
│  │ Solo dominio     │            │ Coordina capas   │       │
│  └──────────────────┘            └──────────────────┘       │
│                                                              │
│  CONOCE:                         CONOCE:                    │
│  - Entidades                     - Servicios de dominio     │
│  - Value Objects                 - Repositorios             │
│  - Otros servicios dominio       - Publicador de eventos    │
│                                  - Servicios externos       │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Uso en Casos de Uso

class ConfirmarPedidoHandler:
    """Caso de uso que orquesta servicios de dominio"""

    def __init__(
        self,
        uow: "UnitOfWork",
        servicio_disponibilidad: ServicioDisponibilidad,
        servicio_precios: ServicioPrecios,
    ):
        self._uow = uow
        self._disponibilidad = servicio_disponibilidad
        self._precios = servicio_precios

    def ejecutar(self, comando: "ConfirmarPedidoCommand") -> None:
        with self._uow:
            pedido = self._uow.pedidos.obtener(comando.pedido_id)
            almacenes = self._uow.almacenes.obtener_activos()

            # Usar servicio de dominio para verificar disponibilidad
            resultados = self._disponibilidad.verificar_disponibilidad_pedido(
                pedido, almacenes
            )

            if not self._disponibilidad.pedido_completamente_disponible(resultados):
                raise StockInsuficiente(resultados)

            pedido.confirmar()
            self._uow.pedidos.guardar(pedido)
            self._uow.commit()

Antipatrones a Evitar

1. Servicio Anémico con Lógica en Otro Lugar

# MAL: Servicio sin lógica real
class ServicioTransferenciasMal:
    def transferir(self, origen_id, destino_id, monto):
        # Solo pasa datos, no tiene lógica
        return {"origen": origen_id, "destino": destino_id}

# BIEN: Servicio con lógica de negocio completa
class ServicioTransferenciasBien:
    def transferir(self, origen: Cuenta, destino: Cuenta, monto: Dinero):
        # Valida, ejecuta, retorna resultado
        pass

2. Servicio con Estado

# MAL: Servicio con estado interno
class ServicioPreciosMal:
    def __init__(self):
        self.ultimo_calculo = None  # Estado!

    def calcular(self, producto):
        self.ultimo_calculo = producto.precio  # Guarda estado
        return self.ultimo_calculo

# BIEN: Sin estado
class ServicioPreciosBien:
    def calcular(self, producto, cliente, cantidad):
        return producto.precio * cantidad  # Puro

3. Servicio que Accede a Infraestructura

# MAL: Servicio accede a BD directamente
class ServicioDisponibilidadMal:
    def __init__(self, db_connection):
        self._db = db_connection  # Infraestructura!

    def verificar(self, producto_id):
        return self._db.query("SELECT stock FROM...")

# BIEN: Recibe datos como parámetros
class ServicioDisponibilidadBien:
    def verificar(self, producto: Producto, almacenes: list[Almacen]):
        # Solo lógica de dominio
        pass

Checklist de Servicios de Dominio