Опирается на правила: R-MOD-1, R-MOD-2, R-AGG-5, R-REP-1, R-REP-2 из DDD Tactical Style Guide → раздел 9. Module (структура пакетов).

Важно знать

  • Верхний уровень core/ — Bounded Context: core/order/, core/customer/, core/product/. Никаких core/entity/, core/service/, core/repository/.
  • Внутри каждого BC: aggregate/, entity/, value_object/, event/, port/, и опционально service/, specification/, factory/.
  • usecase/ живёт рядом с доменными поддиректориями, в том же BC. Там — command/ (мутирующие операции) и query/ (запросы на чтение).
  • adapters/ — отдельная иерархия: adapters/in/http/ и adapters/out/persistence/. Домен про них ничего не знает.
  • core/ не импортирует FastAPI, SQLAlchemy, Pydantic. Только stdlib + core/shared/building_blocks.py. Инвариант проверяется import-linter.
  • port/ содержит Protocol-интерфейсы репозиториев и внешних систем. Реализации — исключительно в adapters/out/.
  • Ссылки между агрегатами — только по ID (R-AGG-5). Импортировать Customer из core/order/ нельзя — только CustomerId.

Структура пакетов выглядит мелочью, но именно по ней через год определяется, был ли в проекте DDD или просто ярлык. Группировка по типу (entity/, service/, repository/) выглядит аккуратно в момент создания, но через 50 классов разработчик ищет «всё про Order» в шести разных папках. Группировка по домену даёт обратное: открыл core/order/, видишь всё, что есть про заказ.

Канонический layout

R-MOD-1: верхний уровень в core/ — Bounded Context. Каждый BC — самодостаточная единица.

core/
  shared/
    building_blocks.py          # Entity, AggregateRoot, DomainEvent
  order/                        # Bounded Context: Order
    aggregate/
      order.py                  # class Order(AggregateRoot[OrderId])
    entity/
      order_line.py             # class OrderLine(Entity[UUID])
    value_object/
      money.py                  # @dataclass(frozen=True) class Money
      order_id.py               # @dataclass(frozen=True) class OrderId
      order_status.py           # enum OrderStatus
    event/
      order_created.py          # @dataclass(frozen=True) class OrderCreated(DomainEvent)
      order_confirmed.py
      order_cancelled.py
    port/
      order_repository.py       # class OrderRepository(Protocol)
    service/                    # опционально
      pricing_service.py
    specification/              # опционально
      eligible_for_refund.py
    usecase/
      command/
        create_order.py         # @dataclass(frozen=True) class CreateOrder
        create_order_handler.py # class CreateOrderHandler
        confirm_order.py
        confirm_order_handler.py
      query/
        get_order.py
        get_order_handler.py
        find_active_orders.py
        find_active_orders_handler.py
  customer/                     # Bounded Context: Customer
    aggregate/...
    value_object/...
    port/...
    usecase/...
  product/                      # Bounded Context: Product
    aggregate/...
    value_object/...
    port/...
    usecase/...

adapters/
  in/
    http/
      order_router.py           # FastAPI router
      order_mapper.py           # HTTP DTO ↔ Command/Query
  out/
    persistence/
      sqlalchemy_order_repository.py   # реализация OrderRepository
      order_record_mapper.py           # ORM-модель ↔ домен

app/
  container.py                  # DI-состав, wire adapters → handlers
  main.py                       # FastAPI app, include_router

Что даёт такая структура:

  • Перенос BC в отдельный сервис. Когда Customer растёт, копируется core/customer/ целиком. Ссылки на другие BC уже идут по ID (R-AGG-5), импорты не размазаны по всему проекту.
  • Чтение «всё про Order» — одна директория. Открыл core/order/, видишь агрегат, события, порт репозитория, use-case-ы. Не нужно искать OrderService в service/, OrderRepository в repository/, Order в entity/.
  • import-linter как CI-барьер. Правило layers: core < adapters < app делает нарушение R-MOD-2 ошибкой сборки, а не замечанием на ревью.

Группировка по типу — антипаттерн

R-MOD-1 явно запрещает:

# ПЛОХО — group-by-layer
src/
  entity/
    order.py
    customer.py
    product.py
  repository/
    order_repository.py
    customer_repository.py
  service/
    order_service.py
    pricing_service.py
  router/
    order_router.py
    customer_router.py

Что не так:

  • Зависимости размазаны. Чтобы понять «как работает Order», открываешь entity/, repository/, service/, router/.
  • Нарушения R-AGG-5 не видны. Импорт entity.customer.Customer в service.order_service выглядит формально нормально — но это ссылка между агрегатами по объекту, не по ID. Структура по типу это скрывает.
  • Рефакторинг BC = операция по всему проекту. Перенос Customer в отдельный сервис потребует вытащить файлы из четырёх разных директорий.

Группировка по типу — наследие старых MVC-фреймворков (models/, views/, controllers/). В DDD от неё отказываются: всё, что связано одним доменом, должно быть рядом.

core/ без FastAPI, SQLAlchemy, Pydantic

R-MOD-2: модули core/<bc>/ не импортируют инфраструктурный стек.

Что разрешено в импортах внутри core/:

  • stdlib — dataclasses, datetime, uuid, decimal, enum, typing.
  • core/shared/building_blocks.py — базовые типы Entity, AggregateRoot, DomainEvent.
  • Другие domain-типы того же BC или идентификаторы соседнего BC.

Что не допускается:

  • fastapi — ни APIRouter, ни Depends, ни Pydantic-схемы.
  • sqlalchemy — ни ORM-модели, ни Session, ни AsyncSession.
  • pydantic — BaseModel, Field, validator не в core/.
  • alembic — скрипты миграций не в core/.
# ПЛОХО — SQLAlchemy-модель в домене
from sqlalchemy import Column, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Order(Base):
    __tablename__ = "orders"
    id = Column(String, primary_key=True)
    status = Column(String)


# ХОРОШО — чистый домен
from core.shared.building_blocks import AggregateRoot
from core.order.value_object.order_id import OrderId
from core.order.value_object.order_status import OrderStatus

class Order(AggregateRoot[OrderId]):
    def __init__(self, id_: OrderId, customer_id: CustomerId) -> None:
        super().__init__(id_)
        self._status = OrderStatus.NEW
        self._lines: list[OrderLine] = []

Маппинг доменного Order ↔ SQLAlchemy-записи делается в adapters/out/persistence/order_record_mapper.py. Домен об этом ничего не знает.

Protocol-порты в port/

R-REP-1: порт репозитория — Protocol в core/<bc>/port/. Реализация — в adapters/out/persistence/ (R-REP-2).

# core/order/port/order_repository.py
from typing import Protocol
from core.order.aggregate.order import Order
from core.order.value_object.order_id import OrderId
from core.customer.value_object.customer_id import CustomerId


class OrderRepository(Protocol):
    async def by_id(self, order_id: OrderId) -> Order | None: ...
    async def save(self, order: Order) -> None: ...
    async def active_by_customer(self, customer_id: CustomerId) -> list[Order]: ...

# adapters/out/persistence/sqlalchemy_order_repository.py
from sqlalchemy.ext.asyncio import AsyncSession
from core.order.port.order_repository import OrderRepository
from core.order.aggregate.order import Order
from core.order.value_object.order_id import OrderId


class SqlAlchemyOrderRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def by_id(self, order_id: OrderId) -> Order | None:
        row = await self._session.get(OrderRecord, str(order_id.value))
        return _to_domain(row) if row else None

    async def save(self, order: Order) -> None:
        record = _to_record(order)
        await self._session.merge(record)

adapters/ знает о core/, core/ не знает об adapters/ — направление зависимостей однонаправленное.

import-linter — enforce на CI

pyproject.toml:

[tool.importlinter]
root_packages = ["core", "adapters", "app"]

[[tool.importlinter.contracts]]
name = "Domain independence"
type = "layers"
layers = ["app", "adapters", "core"]

Запуск: lint-imports. При нарушении — core/order/aggregate/order.py импортирует sqlalchemy — CI падает.

Дополнительный контракт на запрет cross-BC зависимостей по объекту:

[[tool.importlinter.contracts]]
name = "No cross-BC object references"
type = "forbidden"
source_modules = ["core.order"]
forbidden_modules = ["core.customer.aggregate", "core.customer.entity"]

core.order может импортировать только core.customer.value_object.customer_id — идентификатор, не агрегат.

usecase/ — соседняя половина BC

usecase/ живёт рядом с доменными поддиректориями в том же BC. Здесь — команды и запросы с обработчиками:

# core/order/usecase/command/confirm_order.py
from dataclasses import dataclass
from uuid import UUID


@dataclass(frozen=True)
class ConfirmOrder:
    order_id: UUID


# core/order/usecase/command/confirm_order_handler.py
from core.order.port.order_repository import OrderRepository
from core.order.value_object.order_id import OrderId
from core.shared.clock import Clock


class ConfirmOrderHandler:
    def __init__(self, orders: OrderRepository, clock: Clock) -> None:
        self._orders = orders
        self._clock = clock

    async def handle(self, cmd: ConfirmOrder) -> None:
        order = await self._orders.by_id(OrderId(cmd.order_id))
        if order is None:
            raise OrderNotFoundError(cmd.order_id)
        order.confirm(self._clock)
        await self._orders.save(order)

Хендлер оркеструет: загружает агрегат, вызывает доменный метод, сохраняет. Бизнес-правила живут в Order.confirm(), оркестрация — в хендлере.

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

АнтипаттернПравилоЧто взамен
Верхнеуровневые core/entity/, core/service/, core/repository/R-MOD-1Группировка по BC: core/order/, core/customer/
from sqlalchemy import ... в core/<bc>/R-MOD-2SQLAlchemy только в adapters/out/persistence/
from fastapi import ... в core/<bc>/R-MOD-2FastAPI только в adapters/in/http/
from pydantic import BaseModel в core/<bc>/R-MOD-2@dataclass(frozen=True) для VO и событий
Импорт агрегата другого BC: from core.customer.aggregate.customer import CustomerR-AGG-5 + R-MOD-1Только from core.customer.value_object.customer_id import CustomerId
Реализация OrderRepository в core/order/R-REP-2Реализация в adapters/out/persistence/

Куда дальше

  • python/aggregate-root.md — AggregateRoot[ID], регистрация событий, инварианты.
  • python/value-object.md — @dataclass(frozen=True), __post_init__, Decimal для денег.
  • python/domain-event.md — DomainEvent, иммутабельность, публикация после сохранения.
  • python/repository.md — Protocol-порт в core/, реализация в adapters/out/.
  • python/entity.md — Entity[ID], identity-equality, конструктор без сеттеров.
  • python/domain-service.md — когда Domain Service оправдан (≥ 2 агрегатов).
  • python/factory.md — @classmethod create(...), начальные события.
  • python/specification.md — is_satisfied_by, когда вводить спецификацию.
  • Hexagonal Architecture → Python — как core/ вписывается в гексагональную структуру и import-linter.