Опирается на правила: R-FAC-1, R-FAC-2, R-FAC-X1 из DDD Tactical Rules → раздел 7. Factory; смежно R-AGG-3, R-AGG-5, R-EVT-X3.

Важно знать

  • Дефолт — конструктор агрегата. Order(id_=..., customer_id=...) справляется в большинстве случаев; Factory появляется только по триггеру (R-FAC-1).
  • Три триггера: (1) валидация требует другого агрегата, (2) выбор подтипа по политике, (3) сборка из нескольких частей.
  • В Python предпочтительный вариант — @classmethod create(...) прямо на агрегате; отдельный класс фабрики появляется, когда правила усложняются или появляются зависимости.
  • Factory принимает агрегаты параметрами, сам ничего не загружает из репозитория — загрузка в UseCaseHandler.
  • R-FAC-2: фабрика возвращает уже валидный агрегат с зарегистрированными начальными событиями (OrderCreated).
  • Factory живёт в core/<bc>/aggregate/ (classmethod) или core/<bc>/factory/ (отдельный класс) — без FastAPI, SQLAlchemy, Pydantic.
  • R-FAC-X1: Factory ради Factory — антипаттерн. Если конструктор справляется, вводить create() не нужно.

Factory — паттерн, который в Python вводится ещё реже, чем в Java: @dataclass-агрегаты с __post_init__ часто покрывают потребность сами. Но когда «создание» — самостоятельная бизнес-операция с правилами и зависимостями — classmethod или отдельный класс дают чёткое место для этих правил.

Когда вводим

R-FAC-1: один из трёх триггеров должен сработать.

Триггер 1 — валидация требует другого агрегата.

Конструктор Order не может проверить customer.is_active — у него нет ссылки на Customer (R-AGG-5). Factory принимает Customer параметром, проверяет правило, передаёт в агрегат только CustomerId.

# core/order/aggregate/order.py

class Order(AggregateRoot[OrderId]):
    def __init__(self, id_: OrderId, customer_id: CustomerId) -> None:
        super().__init__(id_)
        self._customer_id = customer_id
        self._status = OrderStatus.NEW
        self._lines: list[OrderLine] = []
        self._register_event(
            OrderCreated(uuid7(), datetime.now(UTC), id_.value, customer_id.value)
        )

    @classmethod
    def create_for(
        cls,
        customer: Customer,
        ids: IdGenerator,
        clock: Clock,
    ) -> "Order":
        if not customer.is_active:
            raise CustomerNotActiveError(customer.id)
        if customer.credit_limit < Money(Decimal("0"), "RUB"):
            raise CreditLimitExceededError(customer.id, customer.credit_limit)
        return cls(OrderId(ids.next()), customer.id, clock)

Конструктор регистрирует OrderCreated сам (R-EVT-X3); create_for только применяет бизнес-правила и делегирует вниз.

Триггер 2 — выбор подтипа по политике.

# core/payment/factory/payment_factory.py
from core.payment.aggregate.card_payment import CardPayment
from core.payment.aggregate.sbp_payment import SbpPayment
from core.payment.aggregate.cash_payment import CashPayment
from core.payment.value_object.payment_method import PaymentMethod, PaymentMethodType


class PaymentFactory:
    def create_for(
        self,
        order_id: OrderId,
        method: PaymentMethod,
        amount: Money,
        ids: IdGenerator,
        clock: Clock,
    ) -> Payment:
        payment_id = PaymentId(ids.next())
        match method.type:
            case PaymentMethodType.CARD:
                return CardPayment(payment_id, order_id, method.card_token, amount, clock)
            case PaymentMethodType.SBP:
                return SbpPayment(payment_id, order_id, method.phone, amount, clock)
            case PaymentMethodType.CASH:
                return CashPayment(payment_id, order_id, amount, clock)

Полиморфные агрегаты — редкий случай, но когда они есть (разные типы платежей в Сбере/маркетплейсе), Factory скрывает выбор конкретного подкласса и делает UseCaseHandler независимым от этого выбора.

Триггер 3 — сборка из нескольких частей.

# core/invoice/factory/invoice_factory.py

class InvoiceFactory:
    def create_from_orders(
        self,
        customer: Customer,
        orders: list[Order],
        period: BillingPeriod,
        ids: IdGenerator,
        clock: Clock,
    ) -> Invoice:
        delivered = [o for o in orders if o.status == OrderStatus.DELIVERED]
        if not delivered:
            raise NoOrdersToInvoiceError(customer.id, period)
        lines = tuple(
            InvoiceLine(o.id, o.total(), o.delivered_at)
            for o in delivered
        )
        return Invoice(InvoiceId(ids.next()), customer.id, period, lines, clock)

Invoice собирается из нескольких Order-ов с фильтрацией. Если бы это делал конструктор Invoice, ему пришлось бы принимать list[Order] — нарушение R-AGG-5 (ссылки между агрегатами только по id).

Возвращает валидный агрегат + начальные события

R-FAC-2: на выходе из Factory — агрегат готов к save. Все инварианты проверены, все начальные события зарегистрированы.

# core/order/aggregate/order.py (фрагмент конструктора)

def __init__(self, id_: OrderId, customer_id: CustomerId, clock: Clock) -> None:
    super().__init__(id_)
    self._customer_id = customer_id
    self._status = OrderStatus.NEW
    self._lines: list[OrderLine] = []
    self._register_event(
        OrderCreated(
            event_id=uuid7(),
            occurred_at=clock.now(),
            aggregate_id=id_.value,
            customer_id=customer_id.value,
        )
    )

OrderCreated регистрируется внутри конструктора агрегата — Factory не трогает _register_event напрямую. Это соответствует R-EVT-X3 и сохраняет инвариант: событие возникает в момент изменения состояния, а не снаружи.

Поток вызова в UseCaseHandler:

# core/order/usecase/create_order.py

class CreateOrderHandler:
    def __init__(
        self,
        customers: CustomerRepository,
        orders: OrderRepository,
    ) -> None:
        self._customers = customers
        self._orders = orders

    async def handle(self, command: CreateOrder) -> OrderId:
        customer = await self._customers.by_id(command.customer_id)
        if customer is None:
            raise CustomerNotFoundError(command.customer_id)
        order = Order.create_for(customer, command.ids, command.clock)
        await self._orders.save(order)
        return order.id

Handler не «создаёт» агрегат вручную, не регистрирует событие, не выставляет статус. create_for возвращает готовый агрегат, save публикует события через UoW. Чистый поток.

@classmethod vs отдельный класс

В Python два равноправных варианта:

@classmethod на агрегате — когда правила простые и зависимостей нет:

# core/product/aggregate/product.py

class Product(AggregateRoot[ProductId]):
    @classmethod
    def create(
        cls,
        name: str,
        price: Money,
        ids: IdGenerator,
        clock: Clock,
    ) -> "Product":
        if not name.strip():
            raise ValueError("name must not be blank")
        return cls(ProductId(ids.next()), name, price, clock)

Отдельный класс в core/<bc>/factory/ — когда правила усложняются или нужны зависимости:

# core/order/factory/order_factory.py

class OrderFactory:
    def __init__(self, promo_service: PromoCodeService) -> None:
        self._promo = promo_service

    def create_for(
        self,
        customer: Customer,
        items: list[OrderItem],
        promo_code: str | None,
        ids: IdGenerator,
        clock: Clock,
    ) -> Order:
        if not customer.is_active:
            raise CustomerNotActiveError(customer.id)
        discount = (
            self._promo.resolve(promo_code, customer) if promo_code else Discount.none()
        )
        return Order(OrderId(ids.next()), customer.id, items, discount, clock)

Когда PromoCodeService появился в зависимостях — @classmethod больше не подходит (у classmethod нет self-состояния). Выносим в класс, регистрируем в DI-контейнере:

# app/di.py  (composition root)

order_factory = OrderFactory(promo_service=promo_service)
create_order_handler = CreateOrderHandler(
    customers=customer_repo,
    orders=order_repo,
    factory=order_factory,
)

Factory живёт в core/, не в адаптере

Factory — часть домена (R-FAC-1 подразумевает бизнес-правила). Правила расположения:

core/
  order/
    aggregate/
      order.py          # @classmethod create_for — простой случай
    factory/
      order_factory.py  # отдельный класс — когда нужны зависимости
    port/
      order_repository.py
adapters/
  out/persistence/
    order_repo_impl.py

core/ не импортирует FastAPI, SQLAlchemy, Pydantic — только stdlib (enforce через import-linter, R-MOD-2).

Factory ≠ Builder

Builder — про удобство конструирования (fluent API, опциональные параметры). Factory — про бизнес-правила создания.

# Builder — удобство в тестах
order = (
    OrderBuilder()
    .with_id(OrderId(uuid4()))
    .with_customer(CustomerId(uuid4()))
    .with_line("p-1", qty=2, price=Money(Decimal("100"), "RUB"))
    .build()
)

# Factory — бизнес-правила
order = Order.create_for(customer=customer, ids=ids, clock=clock)

Builder уместен в тестах (fixtures.py, builders.py) и иногда в core, когда у агрегата много опциональных полей. Builder не заменяет Factory: правило «нельзя создать Order для неактивного Customer» нельзя положить в .build() без зависимости от Customer.

Как Factory не делает

R-FAC-X1: Factory ради Factory — антипаттерн.

# ПЛОХО — просто обёртка над конструктором без бизнес-правил
class OrderFactory:
    def create(self, id_: OrderId, customer_id: CustomerId) -> Order:
        return Order(id_=id_, customer_id=customer_id)

Бесполезный слой: дублирует конструктор, правил нет, тестировать нечего.

Распространённые ложные обоснования:

  • «Чтобы скрыть прямой вызов конструктора» — в Python это нормально, скрывать без причины не нужно.
  • «Чтобы везде единообразно через Factory» — cargo-cult; единообразие там, где есть смысл.
  • «Чтобы мокать в тестах» — тест строит агрегат прямым конструктором или через OrderBuilder, mock-фабрика не нужна.

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

АнтипаттернПравилоЧто взамен
Factory ради Factory — обёртка над конструктором без правилR-FAC-X1Order(id_=..., customer_id=...) напрямую
Factory загружает агрегаты из репозиторияR-FAC-1Загрузка в UseCaseHandler, Factory принимает объекты параметрами
Factory регистрирует события напрямуюR-FAC-2, R-EVT-X3События регистрирует конструктор агрегата через _register_event
Factory в adapters/ или с импортом SQLAlchemy/FastAPIR-MOD-2Только core/, без фреймворков
Builder вместо Factory, когда есть бизнес-правила создания—Builder для удобства, Factory для правил

Куда дальше

  • python/aggregate-root.md — _register_event и начальные события в конструкторе.
  • python/domain-event.md — OrderCreated и публикация через UoW после save.
  • python/domain-service.md — соседний паттерн «логика про два агрегата»; часто путают с Factory.
  • python/entity.md — Entity[ID] и identity-equality как основа для агрегата.
  • python/value-object.md — Money, CustomerId и другие VO, которые Factory принимает параметрами.
  • python/repository.md — Protocol-порты и save, который публикует события Factory-агрегата.
  • python/specification.md — отдельный паттерн для правил применимости; не путать с Factory.
  • python/module-structure.md — где именно лежат factory/ и aggregate/ в core/<bc>/.
  • Смежный раздел: Hexagonal Architecture → Python — как DI-контейнер в app/ собирает Factory без фреймворковой магии.