Domain Services
Lógica de dominio que no pertenece a ninguna entidad ni Value Object.
Un Domain Service encapsula lógica de negocio que involucra múltiples aggregates, entidades o Value Objects, y que no pertenece naturalmente a ninguno de ellos. Su existencia está justificada solo cuando la operación carece de un 'hogar' claro en el modelo.
La característica más importante de un Domain Service es que no tiene estado. No guarda datos entre llamadas, no tiene atributos de instancia que representen estado de negocio. Cada invocación es autocontenida. Si descubres que tu 'service' acumula estado, probablemente debería ser una Entity o un Aggregate.
El nombre de un Domain Service debe provenir del lenguaje ubicuo. Debe sonar a operación de dominio, no a un componente técnico. `<SomethingCalculator>`, `<TransferPolicy>`, `<EligibilityChecker>` son buenos nombres. `<DataProcessor>` o `<BusinessLogicHelper>` son señales de que algo está mal.
Los Domain Services viven en la capa de dominio. No dependen de repositorios, bases de datos ni frameworks. Si necesitan acceder a datos, los reciben como parámetros —ya resueltos por la capa de aplicación— nunca los buscan ellos mismos.
En Python idiomático, un Domain Service puede implementarse como una clase con un único método público (generalmente `execute` o un verbo del dominio), o directamente como una función. La clase es preferible cuando el service tiene múltiples operaciones relacionadas o cuando se necesita inyectar collaborators (como políticas).
No confundir Domain Service con Application Service. El Application Service orquesta: llama al repositorio, llama al Domain Service, persiste. El Domain Service calcula o valida usando solo objetos de dominio que ya están en memoria.
# domain/services/<domain_service>.py
# Domain Service: sin estado, nombre del lenguaje ubicuo.
# NO inyectar repositorios, NO acceder a base de datos, NO tener atributos de estado.
from dataclasses import dataclass
from .<model>.<aggregate> import <Aggregate>
from .<model>.<value_object> import <ValueObject>
from ..errors import <DomainError>
# Opción A: clase sin estado (recomendada si hay múltiples operaciones relacionadas)
class <DomainServiceName>:
"""
Encapsula la lógica de <business_operation>.
Sin estado: no guarda datos entre llamadas.
"""
def <execute>(
self,
<aggregate_a>: <AggregateA>,
<aggregate_b>: <AggregateB>,
<policy>: <PolicyValueObject>,
) -> <ResultType>:
# Toda la lógica usa objetos que llegan como parámetros.
# NO llamar a repositorios desde aquí.
if not self._<validate_rule>(<aggregate_a>, <policy>):
raise <DomainError>('<business_reason>')
return self._<compute_result>(<aggregate_a>, <aggregate_b>)
def _<validate_rule>(
self, <aggregate>: <AggregateA>, <policy>: <PolicyValueObject>
) -> bool:
# Regla privada de validación — solo lógica pura, sin I/O
...
# Opción B: función libre (para operaciones sencillas y aisladas)
def <domain_operation>(
<aggregate_a>: <AggregateA>,
<aggregate_b>: <AggregateB>,
) -> <ValueObject>:
"""Calcula <result> a partir de los aggregates. Sin efectos secundarios."""
...
# application/commands/<do_something>/handler.py (fragmento)
# La capa de aplicación resuelve los aggregates y llama al Domain Service.
class <DoSomethingHandler>:
def __init__(self, repo_a: <AggregateARepository>, svc: <DomainServiceName>) -> None:
self._repo_a = repo_a
self._svc = svc
def handle(self, cmd: <DoSomethingCommand>) -> None:
a = self._repo_a.get_by_id(cmd.<aggregate_a_id>) # La app resuelve
b = ...
result = self._svc.<execute>(a, b, cmd.<policy>) # Domain Service calcula
self._repo_a.save(a)Debugging lab
Detecta y corrige el error o la violación de diseño.
- 4.2.5.1
class <TransferService>: def __init__(self, repo: <AccountRepository>) -> None: self._repo = repo # ← error: Domain Service con repositorio inyectado def transfer(self, from_id: UUID, to_id: UUID, amount: Money) -> None: source = self._repo.get(from_id) target = self._repo.get(to_id) source.debit(amount) target.credit(amount)
- 4.2.5.2
class <PricingService>: _last_discount: Decimal = Decimal('0') # ← estado entre llamadas def calculate(self, order: <Order>, policy: <DiscountPolicy>) -> Money: self._last_discount = policy.discount_rate return order.subtotal * (1 - self._last_discount)
- 4.2.5.3
# application/commands/approve_loan/handler.py class ApproveLoanHandler: def handle(self, cmd: ApproveLoanCommand) -> None: applicant = self._repo.get(cmd.applicant_id) # ← La lógica de elegibilidad está directamente aquí, en el handler if applicant.credit_score < 650 or applicant.income < cmd.amount * 3: raise ApplicationError('Not eligible') applicant.approve_loan(cmd.amount)
- 4.2.5.4
# domain/services/exchange_rate_service.py class ExchangeRateService: async def get_rate(self, from_currency: str, to_currency: str) -> Decimal: response = await httpx.get(f'https://api.rates.io/{from_currency}/{to_currency}') # ← I/O en dominio return Decimal(response.json()['rate'])
- 4.2.5.5
class <InventoryManager>: # ← nombre genérico def process_data(self, items: list[dict]) -> dict: # ← trabaja con dicts crudos result = {} for item in items: if item['stock'] < item['min_stock']: result[item['id']] = 'reorder' return result