Domain Service в DDD на Python — stateless-класс для логики между агрегатами

Опирается на правила: R-DS-1 … R-DS-3 и R-DS-X1 … R-DS-X2 из DDD Tactical Style Guide → раздел 6. Domain Service.

Важно знать

• Domain Service вводится только если логика затрагивает два и более агрегата и не вписывается в один корень (R-DS-1). Правило для одного агрегата — метод агрегата.
• Класс stateless: нет полей с изменяемым состоянием. Принимает доменные объекты (Account, Order), не DTO, не репозитории, не сессии (R-DS-2).
• Имя выражает доменную операцию: TransferService, PricingService, FraudDetectionService (R-DS-3). Не OrderHelper, не BusinessLogicService.
• Оркестрация остаётся в Handler: загрузка агрегатов из репозитория, транзакции, публикация событий — всё это Handler, не Domain Service (R-DS-X1).
• Domain Service — не свалка для всей логики, которую лень класть в агрегат. Агрегаты должны быть богатыми (R-DS-X2).
• В Python Domain Service — обычный класс (не dataclass, не Protocol). Зависимости через __init__ только если это другие Domain Service-ы или pure-domain вспомогательные объекты.

Domain Service — самый редкий тактический паттерн. В типичном сервисе 1–3 Domain Service-а, не 20. Главная проверка перед введением: «эта логика принадлежит одному агрегату или двум?». Если одному — это метод агрегата. Если двум — рассмотри Domain Service. Раскрытие раздела 6 гайда.

Когда вводим

R-DS-1: Domain Service нужен, когда логика касается нескольких агрегатов и не может быть выражена методом одного из них.

Пример — перевод средств между счетами. Логика «снять с одного, положить на другой» касается двух Account. Ни один из счетов не должен знать про другой:

# core/account/service/transfer_service.py
from core.account.aggregate.account import Account
from core.account.value_object.money import Money


class TransferService:
    def transfer(self, source: Account, destination: Account, amount: Money) -> None:
        source.withdraw(amount)
        destination.deposit(amount)

Оба агрегата остаются инкапсулированными — Account.withdraw и Account.deposit реализуют своё поведение и регистрируют события. TransferService лишь координирует вызовы в правильном порядке.

Ещё один пример — расчёт цены с учётом клиентских скидок:

# core/order/service/pricing_service.py
from decimal import Decimal

from core.order.aggregate.order import Order
from core.customer.aggregate.customer import Customer
from core.order.value_object.money import Money


class PricingService:
    def calculate_discount(self, order: Order, customer: Customer) -> Decimal:
        base = order.subtotal().amount
        if customer.is_vip() and base > Decimal("10000"):
            return Decimal("0.15")
        if customer.loyalty_years() >= 3:
            return Decimal("0.07")
        return Decimal("0")

PricingService получает готовые агрегаты — не загружает их из репозитория сам.

Stateless — без изменяемого состояния

R-DS-2: Domain Service stateless. Принимает доменные объекты, возвращает результат. Никаких self._cache, self._counter, self._session.

# ХОРОШО — stateless
class TransferService:
    def transfer(self, source: Account, destination: Account, amount: Money) -> None:
        source.withdraw(amount)
        destination.deposit(amount)

# ПЛОХО — stateful Domain Service
class TransferService:
    def __init__(self) -> None:
        self._transfer_count = 0   # состояние в Domain Service — R-DS-2 нарушено

    def transfer(self, source: Account, destination: Account, amount: Money) -> None:
        self._transfer_count += 1
        ...

Если нужен счётчик переводов — это метрика приложения (Application layer), не доменное состояние.

Имя — доменная операция

R-DS-3: имя класса выражает доменную операцию, не инфраструктурный или технический концепт.

# ХОРОШО
class TransferService: ...           # операция «перевод»
class PricingService: ...            # операция «расчёт цены»
class FraudDetectionService: ...     # операция «обнаружение мошенничества»
class ReservationService: ...        # операция «резервирование»

# ПЛОХО
class OrderHelper: ...               # «помощник» — без семантики
class BusinessLogicService: ...      # тавтология
class OrderManager: ...              # «менеджер» — слишком общо
class DataProcessingService: ...     # технический термин

Оркестрация остаётся в Handler

R-DS-X1: загрузка агрегатов из репозитория, управление транзакцией, публикация событий — в Handler (Application layer), не в Domain Service.

# ПЛОХО — оркестрация в Domain Service
class TransferService:
    def __init__(self, accounts: AccountRepository, session: AsyncSession) -> None:
        self._accounts = accounts     # репозиторий в Domain Service — нарушение
        self._session = session

    async def transfer(self, from_id: AccountId, to_id: AccountId, amount: Money) -> None:
        source = await self._accounts.by_id(from_id)      # загрузка здесь
        dest = await self._accounts.by_id(to_id)
        source.withdraw(amount)
        dest.deposit(amount)
        await self._accounts.save(source)                  # сохранение здесь
        await self._accounts.save(dest)
        await self._session.commit()                       # транзакция здесь

# ХОРОШО — оркестрация в Handler
class TransferFundsHandler:
    def __init__(
        self,
        accounts: AccountRepository,
        transfer_service: TransferService,
    ) -> None:
        self._accounts = accounts
        self._transfer_service = transfer_service

    async def handle(self, cmd: TransferFunds) -> None:
        source = await self._accounts.by_id(cmd.from_account_id)
        if source is None:
            raise AccountNotFoundError(cmd.from_account_id)
        dest = await self._accounts.by_id(cmd.to_account_id)
        if dest is None:
            raise AccountNotFoundError(cmd.to_account_id)

        self._transfer_service.transfer(source, dest, cmd.amount)  # чистый домен

        await self._accounts.save(source)
        await self._accounts.save(dest)

Handler знает про репозиторий и транзакцию. Domain Service знает только про агрегаты.

Domain Service — не свалка

R-DS-X2: если вся доменная логика уходит в Domain Service, агрегаты становятся анемичными — нарушение R-ENT-X5.

# ПЛОХО — Domain Service вместо методов агрегата
class OrderService:
    def confirm_order(self, order: Order) -> None:
        if not order._lines:         # лезем во внутренности
            raise ValueError("empty")
        order._status = OrderStatus.CONFIRMED   # меняем напрямую

# ХОРОШО — логика в методе агрегата
class Order(AggregateRoot[OrderId]):
    def confirm(self, clock: Clock) -> None:
        if not self._lines:
            raise ValueError("cannot confirm empty order")
        self._status = OrderStatus.CONFIRMED
        self._register_event(OrderConfirmed(...))

Правило: если логика принадлежит одному агрегату — это метод агрегата. Domain Service — только когда два и более.

Что запрещено

Куда дальше

• DDD Tactical → раздел 6. Domain Service — нормативные формулировки R-DS-*.
• python/aggregate-root.md — когда логика принадлежит агрегату, а не Domain Service.
• python/specification.md — паттерн для переиспользуемых доменных правил-предикатов.
• python/module-structure.md — где живут Domain Service-ы в структуре core/<bc>/service/.