Specification в DDD — переиспользуемое доменное правило на Python

Опирается на правила: R-SPEC-1, R-SPEC-2, R-SPEC-X1, R-SPEC-X2 из DDD Tactical Style Guide → раздел 8. Specification.

Важно знать

• Specification — доменное правило в самостоятельном объекте: метод is_satisfied_by(candidate) -> bool отвечает на вопрос «соответствует ли объект правилу».
• В Python нет библиотеки ddd-building-blocks. Specification — обычный класс в core/<bc>/specification/, базовый тип ручной: is_satisfied_by как абстрактный метод или Protocol.
• Вводится только когда правило применяется в двух и более местах или требуется комбинация через and/or/not. В остальных случаях — if на агрегате.
• Работает в памяти: принимает Entity / VO, возвращает bool. Не строит SQL-запросы.
• core/<bc>/specification/ не импортирует FastAPI, SQLAlchemy, Pydantic. Модуль домена остаётся чистым Python.
• R-SPEC-X1 Specification, генерирующая WHERE-условие — антипаттерн: ломает изоляцию core/ и путает write-side с read-side.
• R-SPEC-X2 Specification для одного if в одном месте — преждевременная абстракция. Простое правило живёт методом на агрегате.
• Имя — утверждение: EligibleForRefund, VipCustomer, ProductInStock. Суффикс Spec опционален; важно единообразие в проекте.

Specification — компактный паттерн для случаев, когда одно и то же доменное правило нужно проверить в нескольких точках: в command-handler-е при проверке команды, в маппере при формировании view, в доменном сервисе при межагрегатной логике. Размер типичного сервиса — 3–6 спецификаций, не 50.

Базовый тип

В Python базовый класс ручной: два варианта — абстрактный метод и Protocol.

# core/shared/specification.py
from abc import ABC, abstractmethod
from typing import Generic, TypeVar

T = TypeVar("T")


class Specification(ABC, Generic[T]):
    @abstractmethod
    def is_satisfied_by(self, candidate: T) -> bool: ...

    def and_(self, other: "Specification[T]") -> "Specification[T]":
        return _And(self, other)

    def or_(self, other: "Specification[T]") -> "Specification[T]":
        return _Or(self, other)

    def not_(self) -> "Specification[T]":
        return _Not(self)


class _And(Specification[T]):
    def __init__(self, left: Specification[T], right: Specification[T]) -> None:
        self._left = left
        self._right = right

    def is_satisfied_by(self, candidate: T) -> bool:
        return self._left.is_satisfied_by(candidate) and self._right.is_satisfied_by(candidate)


class _Or(Specification[T]):
    def __init__(self, left: Specification[T], right: Specification[T]) -> None:
        self._left = left
        self._right = right

    def is_satisfied_by(self, candidate: T) -> bool:
        return self._left.is_satisfied_by(candidate) or self._right.is_satisfied_by(candidate)


class _Not(Specification[T]):
    def __init__(self, inner: Specification[T]) -> None:
        self._inner = inner

    def is_satisfied_by(self, candidate: T) -> bool:
        return not self._inner.is_satisfied_by(candidate)

Specification живёт в core/shared/ — вне Bounded Context-ов, как Entity и AggregateRoot.

Когда вводим

R-SPEC-2: правило переиспользуется или нужна комбинация.

Правило «заказ подлежит возврату» применяется в двух точках: маппер формирует флаг для UI, handler проверяет перед оформлением. Одна реализация — нет риска расхождения.

# core/order/specification/eligible_for_refund.py
from datetime import timedelta
from core.order.aggregate.order import Order, OrderStatus
from core.shared.clock import Clock
from core.shared.specification import Specification

REFUND_WINDOW = timedelta(days=14)


class EligibleForRefund(Specification[Order]):
    def __init__(self, clock: Clock) -> None:
        self._clock = clock

    def is_satisfied_by(self, order: Order) -> bool:
        if order.status != OrderStatus.DELIVERED:
            return False
        return (self._clock.now() - order.delivered_at) <= REFUND_WINDOW

Применяется в двух местах:

# adapters/in/http/order_mapper.py  (вне core/ — зависимость идёт снаружи)
class OrderViewMapper:
    def __init__(self, refund_spec: EligibleForRefund) -> None:
        self._refund_spec = refund_spec

    def to_view(self, order: Order) -> OrderView:
        return OrderView(
            id=order.id.value,
            total=order.total().amount,
            currency=order.total().currency,
            can_refund=self._refund_spec.is_satisfied_by(order),
        )

# core/order/usecase/issue_refund.py
class IssueRefundHandler:
    def __init__(
        self,
        orders: OrderRepository,
        refund_spec: EligibleForRefund,
        uow: UnitOfWork,
    ) -> None:
        self._orders = orders
        self._refund_spec = refund_spec
        self._uow = uow

    async def handle(self, cmd: IssueRefund) -> RefundId:
        async with self._uow:
            order = await self._orders.by_id(cmd.order_id)
            if order is None:
                raise OrderNotFoundError(cmd.order_id)
            if not self._refund_spec.is_satisfied_by(order):
                raise OrderNotEligibleForRefundError(order.id)
            refund = order.issue_refund(cmd.reason, self._uow.clock)
            await self._orders.save(order)
            return refund.id

Комбинаторы and_/or_/not_

R-SPEC-1: базовый класс даёт and_, or_, not_. Правила собираются из атомарных проверок.

# core/customer/specification/vip_customer.py
from core.customer.aggregate.customer import Customer
from core.order.value_object.money import Money
from core.shared.specification import Specification
from decimal import Decimal


class VipCustomer(Specification[Customer]):
    def __init__(self, threshold: Money) -> None:
        self._threshold = threshold

    def is_satisfied_by(self, customer: Customer) -> bool:
        return customer.total_spent().amount >= self._threshold.amount


class LongtimeCustomer(Specification[Customer]):
    MIN_YEARS = 3

    def is_satisfied_by(self, customer: Customer) -> bool:
        from datetime import datetime
        age = (datetime.utcnow() - customer.registered_at).days // 365
        return age >= self.MIN_YEARS


class HasActiveSubscription(Specification[Customer]):
    def is_satisfied_by(self, customer: Customer) -> bool:
        return customer.active_subscription is not None

Композиция:

vip_spec = VipCustomer(threshold=Money(Decimal("100000"), "RUB"))
longtime_spec = LongtimeCustomer()
subscription_spec = HasActiveSubscription()

premium_discount_spec = vip_spec.or_(longtime_spec.and_(subscription_spec))

if premium_discount_spec.is_satisfied_by(customer):
    order.apply_discount(Decimal("0.15"))

Что это даёт:

• Правило читается как фраза. «VIP или (давний клиент и активная подписка)» — структура прозрачна без чтения тел методов.
• Атомарные спецификации тестируются отдельно. VipCustomer и LongtimeCustomer — независимые unit-тесты; тест на композицию проверяет все комбинации.
• Добавить условие — расширить выражение. Появился статус «сотрудник Сбера» — SberbankStaff spec, or_ в цепочке. Существующий код не меняется.

Без спецификаций то же правило выглядит так:

is_eligible = (
    customer.total_spent().amount >= Decimal("100000")
    or (
        (datetime.utcnow() - customer.registered_at).days // 365 >= 3
        and customer.active_subscription is not None
    )
)

Работает, но: правило размазано по выражению, атомарные части не переиспользовать, добавить четвёртое условие — переписать всё выражение.

Specification ≠ SQL builder

R-SPEC-X1: Specification, строящая SQLAlchemy-условие — антипаттерн.

# ПЛОХО — Specification генерирует SQL
from sqlalchemy import Column, and_

class EligibleForRefundSql:
    def to_clause(self, order_table):
        return and_(
            order_table.c.status == "DELIVERED",
            order_table.c.delivered_at >= datetime.utcnow() - timedelta(days=14),
        )

Что не так:

• core/ тянет SQLAlchemy. Specification в домене начинает зависеть от ORM, нарушается R-MOD-2. import-linter падает.
• Read/write путаются. Repository принимает Specification и через неё строит SELECT. Это уже Query Side, а не Repository.
• In-memory и SQL расходятся. Логика в is_satisfied_by и в to_clause дублируется и живёт несинхронно: правило поменяли в одном месте, второе не заметили.

Правильное разделение:

# core/order/specification/eligible_for_refund.py  — pure domain, in-memory
class EligibleForRefund(Specification[Order]):
    def is_satisfied_by(self, order: Order) -> bool: ...

# adapters/out/persistence/order_query_repository.py  — read-side, SQLAlchemy
class OrderQueryRepository:
    async def find_eligible_for_refund(self, since: datetime) -> list[OrderSummary]:
        stmt = select(order_table).where(
            order_table.c.status == "DELIVERED",
            order_table.c.delivered_at >= since,
        )
        ...

Два класса с похожими критериями — осознанное разделение write-side и read-side. Синхронизацию критериев фиксируем тестом.

Specification для одного места — не нужна

R-SPEC-X2: преждевременная абстракция.

# ПЛОХО — спецификация для одного if
class OrderCancellable(Specification[Order]):
    def is_satisfied_by(self, order: Order) -> bool:
        return order.status != OrderStatus.SHIPPED


class CancelOrderHandler:
    def __init__(self, spec: OrderCancellable, ...) -> None:
        self._spec = spec

    async def handle(self, cmd: CancelOrder) -> None:
        order = await self._orders.by_id(cmd.order_id)
        if not self._spec.is_satisfied_by(order):
            raise OrderNotCancellableError(order.id)
        order.cancel()
        ...

Простое правило в одном месте — метод на агрегате:

# core/order/aggregate/order.py
class Order(AggregateRoot[OrderId]):
    def cancel(self, clock: Clock) -> None:
        if self._status == OrderStatus.SHIPPED:
            raise OrderAlreadyShippedError(self.id)
        self._status = OrderStatus.CANCELLED
        self._register_event(OrderCancelled(uuid7(), clock.now(), self.id.value))

Specification вводится когда правило начинает переиспользоваться, не «на вырост».

Имя — утверждение

R-SPEC-1 подразумевает: имя — утверждение, проверяемое через is_satisfied_by.

# ХОРОШО
EligibleForRefund
VipCustomer
ProductInStock
OrderWithinDeliveryArea

# ПЛОХО
OrderValidator        # validator валидирует входные данные, а не доменные правила
OrderChecker          # непонятно, что именно
RefundRule            # слишком общо
OrderEligibilityHelper  # Helper — категорически нет

Суффикс Spec опционален (EligibleForRefundSpec vs EligibleForRefund). Важно единообразие в проекте.

Callable-предикат как альтернатива

Когда спецификация тривиальна и комбинаторы не нужны — допустим callable-предикат вместо класса:

# core/product/specification/in_stock.py
from core.product.aggregate.product import Product


def in_stock(product: Product) -> bool:
    return product.stock_qty > 0

Применяется напрямую: filter(in_stock, products). Если правило начинает нужно в двух местах или появляются комбинаторы — переводим в класс Specification[Product].

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

Антипаттерн	Правило	Что взамен
Specification строит SQLAlchemy-условие (to_clause())	R-SPEC-X1	Pure-domain is_satisfied_by + отдельный метод в QueryRepository
Specification для одного if в одном месте	R-SPEC-X2	Метод на агрегате или простой if
core/specification/ импортирует FastAPI / SQLAlchemy / Pydantic	R-SPEC-X1, R-MOD-2	Только stdlib + core/shared/building_blocks.py
Имя OrderHelper, Checker, Rule, Validator	R-SPEC-1	Утверждение: EligibleForRefund, VipCustomer

Куда дальше

• python/aggregate-root.md — простые проверки живут на агрегате, а не в Specification.
• python/domain-service.md — соседний паттерн: правило касается ≥ 2 агрегатов и не сводится к bool.
• python/repository.md — read-side фильтрация (Query Repository) как альтернатива Specification для SQL.
• python/value-object.md — VO как носитель ограничений, которые не нужно выносить в Specification.
• python/entity.md, python/domain-event.md, python/factory.md, python/module-structure.md — остальные паттерны раздела.
• DDD Tactical Style Guide — нормативный хаб, раздел 8. Specification.