← назад к разделу

Когда сервис растёт, бизнес-логика начинает расползаться: кусок в роутере FastAPI, кусок в SQLAlchemy-модели, ещё кусок в Kafka-обработчике. Протестировать её без поднятия базы уже нельзя. Поменять роутер на другой фреймворк — страшно. Это классическая боль, которую решает Hexagonal Architecture.

Идея: выделить core — слой, который содержит только бизнес-логику и ни о чём внешнем не знает. Никаких импортов из FastAPI, SQLAlchemy или любой другой библиотеки. Только Python.

Что живёт в core/

core/ — это сердце сервиса. Здесь:

  • агрегаты — доменные объекты с бизнес-логикой (Order, User, Invoice);
  • value objects — неизменяемые типы-значения (Money, CustomerId);
  • доменные события (OrderConfirmedEvent);
  • исключения (OrderNotFoundError, EmptyOrderError);
  • outbound-порты — Protocol-интерфейсы, которые описывают «что core нужно от внешнего мира»;
  • use cases — пары Command/Query + Handler.

Всё это написано на чистом Python. Нет импортов из fastapi, sqlalchemy, pydantic, httpx, aiokafka. Если такой импорт появился в core/ — значит, что-то лежит не там.

Структура папок

Типичная раскладка выглядит так:

src/orders/
  core/
    order/                           # Bounded Context «Orders»
      aggregate/
        order.py                     # Aggregate Root
      entity/
        order_item.py
      value_object/
        money.py
        customer_id.py
      event/
        order_confirmed_event.py
      exception/
        order_not_found_error.py
        empty_order_error.py
      port/
        out/
          order_repository.py        # Protocol — outbound-порт
          payment_port.py
    usecase/
      command/
        create_order_command.py
        confirm_order_command.py
      query/
        get_order_query.py
    service/                         # shared domain-логика (редко)

Папка port/out/ содержит Protocol-интерфейсы: то, что core ожидает от инфраструктуры. Реализуют их адаптеры снаружи core/.

Rich domain против анемичной модели

Частая ошибка — сделать доменный класс просто контейнером полей, а всю логику положить в OrderService. Это называют анемичной моделью, и у неё есть конкретные проблемы:

  • Инварианты расползаются. Логика подтверждения заказа дублируется в роутере, в Kafka-обработчике, в CLI. Одна из копий рано или поздно отстанет.
  • Тест невозможен без инфраструктуры. Order без методов тестируется только через Service с поднятой базой.
  • Жизненный цикл нечитаем. Чтобы понять, когда меняется status, нужно искать по всему коду, а не смотреть в один метод.

Правильный подход — rich domain: бизнес-логика живёт внутри агрегата.

# core/order/aggregate/order.py
from __future__ import annotations
from dataclasses import dataclass, field
from enum import StrEnum
from typing import TYPE_CHECKING

from orders.core.order.value_object.money import Money
from orders.core.order.event.order_confirmed_event import OrderConfirmedEvent
from orders.core.order.exception.empty_order_error import EmptyOrderError
from orders.core.order.exception.invalid_status_error import InvalidStatusError

if TYPE_CHECKING:
    from orders.core.order.entity.order_item import OrderItem


class OrderStatus(StrEnum):
    DRAFT = "DRAFT"
    CONFIRMED = "CONFIRMED"
    CANCELLED = "CANCELLED"


@dataclass
class Order:
    id: str
    customer_id: str
    items: list[OrderItem]
    total: Money
    status: OrderStatus = OrderStatus.DRAFT
    _events: list = field(default_factory=list, repr=False)

    def confirm(self) -> None:
        if not self.items:
            raise EmptyOrderError(self.id)
        if self.status != OrderStatus.DRAFT:
            raise InvalidStatusError(self.status, OrderStatus.DRAFT)
        if self.total <= Money.zero():
            raise ValueError(f"Total must be positive, got {self.total}")
        self.status = OrderStatus.CONFIRMED
        self._events.append(OrderConfirmedEvent(order_id=self.id, total=self.total))

    def cancel(self, reason: str) -> None:
        if self.status == OrderStatus.CANCELLED:
            raise InvalidStatusError(self.status, "not CANCELLED")
        self.status = OrderStatus.CANCELLED

    def pop_events(self) -> list:
        events, self._events = self._events, []
        return events

Метод confirm() — одно место, где описаны все правила подтверждения. Не надо искать по всему проекту.

Use case — тонкий оркестратор

Handler не содержит бизнес-логики. Он берёт агрегат из репозитория, вызывает метод, сохраняет обратно:

# core/usecase/command/confirm_order_command.py
from dataclasses import dataclass
from orders.core.order.port.out.order_repository import OrderRepository


@dataclass(frozen=True)
class ConfirmOrderCommand:
    order_id: str


class ConfirmOrderCommandHandler:

    def __init__(self, order_repository: OrderRepository) -> None:
        self._order_repository = order_repository

    async def handle(self, command: ConfirmOrderCommand) -> None:
        order = await self._order_repository.find_by_id(command.order_id)
        order.confirm()                    # вся логика в агрегате
        await self._order_repository.save(order)

Если в Handler'е появляется условие if ... else с бизнес-смыслом — это сигнал, что логика ушла не туда. Переносим в агрегат.

Outbound-порт как Protocol

core/ не знает, как реализован репозиторий — через SQLAlchemy, через HTTP или in-memory. Он знает только контракт: «дай мне Order по id» и «сохрани Order». Контракт объявляется через Protocol:

# core/order/port/out/order_repository.py
from typing import Protocol
from orders.core.order.aggregate.order import Order


class OrderRepository(Protocol):
    async def find_by_id(self, order_id: str) -> Order: ...
    async def save(self, order: Order) -> None: ...
    async def find_all_by_customer(self, customer_id: str) -> list[Order]: ...

Protocol — структурная типизация. Адаптер реализует его без явного наследования: достаточно совпадения сигнатур. Это упрощает подмену в тестах.

Доменные исключения тоже объявляются в core/:

# core/order/exception/order_not_found_error.py
class OrderNotFoundError(Exception):
    def __init__(self, order_id: str) -> None:
        super().__init__(f"Order {order_id} not found")
        self.order_id = order_id

Handler ловит OrderNotFoundError. Роутер FastAPI преобразует его в HTTP 404. Ни core/, ни Handler не знают про HTTP — это задача адаптера.

Внедрение зависимостей без аннотаций фреймворка

В Python нет Spring-сканирования. Классы в core/ — чистые Python-классы с __init__. Никаких декораторов @injectable или @service не нужно.

Связывание происходит в app/container.py снаружи core/:

# app/container.py
from dependency_injector import containers, providers
from orders.core.usecase.command.confirm_order_command import ConfirmOrderCommandHandler
from orders.adapters.out.persistence.order_sqlalchemy_repository import OrderSQLAlchemyRepository


class Container(containers.DeclarativeContainer):
    order_repository = providers.Singleton(OrderSQLAlchemyRepository)
    confirm_order_handler = providers.Singleton(
        ConfirmOrderCommandHandler,
        order_repository=order_repository,
    )

core/ о Container не знает. Это позволяет в тестах подменить репозиторий на in-memory реализацию без поднятия базы:

# tests/unit/core/test_confirm_order.py
import pytest
from orders.core.order.aggregate.order import Order, OrderStatus
from orders.core.order.value_object.money import Money
from orders.core.usecase.command.confirm_order_command import (
    ConfirmOrderCommand,
    ConfirmOrderCommandHandler,
)


class InMemoryOrderRepository:
    def __init__(self, orders: dict) -> None:
        self._orders = orders

    async def find_by_id(self, order_id: str) -> Order:
        if order_id not in self._orders:
            from orders.core.order.exception.order_not_found_error import OrderNotFoundError
            raise OrderNotFoundError(order_id)
        return self._orders[order_id]

    async def save(self, order: Order) -> None:
        self._orders[order.id] = order


@pytest.mark.asyncio
async def test_confirm_order_sets_status_confirmed():
    order = Order(id="ord-1", customer_id="cust-42", items=[...], total=Money(500, "RUB"))
    repo = InMemoryOrderRepository({"ord-1": order})
    handler = ConfirmOrderCommandHandler(order_repository=repo)

    await handler.handle(ConfirmOrderCommand(order_id="ord-1"))

    assert order.status == OrderStatus.CONFIRMED

Тест запускается без PostgreSQL и без FastAPI — моментально.

Как защитить границы core/ автоматически

В Python самодисциплина не работает на масштабе команды. Случайный from sqlalchemy import ... в core/ никто не заметит на ревью. Для этого есть import-linter.

# pyproject.toml
[tool.importlinter]
root_package = "orders"

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

[[tool.importlinter.contracts]]
name = "core-forbidden-infra"
type = "forbidden"
source_modules = ["orders.core"]
forbidden_modules = [
    "fastapi",
    "sqlalchemy",
    "pydantic",
    "httpx",
    "aiokafka",
]

Запуск в CI:

lint-imports

Если в core/ появится запрещённый импорт — CI упадёт с читаемым сообщением ещё до code review. Pull request не смерджится.

Распространённые ошибки

from fastapi import HTTPException в core/. Доменный слой не должен знать про HTTP. Объявите доменное исключение OrderNotFoundError в core/exception/, а маппинг в HTTPException делайте в роутере.

SQLAlchemy-модель как доменный тип. Если Order — это DeclarativeBase, то core/ уже зависит от SQLAlchemy. Держите ORM-модели в adapters/out/persistence/ и добавляйте маппер между ними и доменным агрегатом.

Pydantic-схема запроса в core/. CreateOrderRequest — это REST-деталь. В core/ лежит CreateOrderCommand (dataclass). Pydantic-схема живёт в adapters/in/http/.

Protocol порта объявлен в адаптере. Порт — контракт core-к-инфраструктуре, он принадлежит core/. Адаптер реализует его, но не владеет им.

Коротко

  • core/ зависит только от стандартной библиотеки Python. Никакого FastAPI, SQLAlchemy, Pydantic.
  • Бизнес-логика живёт внутри агрегата (order.confirm()), а не в сервисном классе.
  • Outbound-порт — это Protocol в core/<bc>/port/out/. Адаптер реализует его снаружи.
  • Handler — тонкий оркестратор: взял агрегат, вызвал метод, сохранил.
  • Классы core/ — чистые Python-классы без DI-аннотаций. Связывание происходит в app/container.py.
  • import-linter в CI — единственный надёжный способ защитить границы в Python.
  • Тесты на core/ запускаются без базы и без HTTP — быстро и изолированно.

Что почитать дальше