Core слой в Hexagonal на Python — агрегаты, ports, use cases без FastAPI и SQLAlchemy

Core слой Hexagonal на Python: агрегаты, Protocol-порты, use cases без FastAPI и SQLAlchemy. Rich domain vs анемичная модель. Контракт import-linter. Правила R-HEX-CORE-1..4.

Опубликовано 20 июня 2026 ~6 мин чтения Автор: Вадим Викулин

Опирается на правила: R-HEX-CORE-1 … R-HEX-CORE-4 и R-HEX-CORE-X1 … R-HEX-CORE-X5 из Hexagonal Style Guide → раздел 3. Core слой.

Важно знать

core/ зависит только от stdlib Python. Никаких fastapi, sqlalchemy, pydantic, httpx, aiokafka — это инфраструктура.

Outbound-порт объявляется как Protocol (или ABC) в core/<bc>/port/out/. Реализует его out-adapter, не core/.

Rich domain: бизнес-логика внутри агрегата (order.confirm()), не в *Service-классах. Анемичная модель — антипаттерн.

Pydantic BaseModel (CreateOrderRequest) в core/ — запрещена. REST-схемы — деталь adapters/in/http/.

SQLAlchemy-модель (Base, DeclarativeBase) в core/ как доменный тип — запрещена. Это деталь adapters/out/persistence/.

Границы core/ охраняет import-linter контракт layers в pyproject.toml — единственный надёжный guard в Python.

DI-аннотации и фреймворк-специфичные декораторы в core/ не нужны: объекты — чистые Python-классы, wiring — в app/container.py.

core/ — это сердце сервиса. Здесь живут агрегаты, бизнес-правила, инварианты, события. Это то, что не зависит ни от какой инфраструктуры и могло бы работать без FastAPI, без PostgreSQL, без HTTP. На практике мы запускаем это в FastAPI, но сам core этого не знает — он принимает зависимости через port-Protocol'ы, которые реализуют адаптеры. Раскрытие правил R-HEX-CORE-* ниже.

Что разрешено в core/

R-HEX-CORE-1: core/ зависит только от stdlib.

# pyproject.toml — dependencies core
# core/ не добавляет сторонних пакетов.
# Единственное допустимое исключение — внутренние DDD-хелперы,
# если они сами не тянут инфраструктурных зависимостей.
dependencies = []

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

fastapi, starlette — web-фреймворк.
sqlalchemy, asyncpg, psycopg — persistence.
pydantic — REST-сериализация. Dataclasses или голые Python-классы для domain-типов.
httpx, aiohttp — HTTP-клиенты к внешним системам.
aiokafka, confluent-kafka — Kafka.
Любая другая инфраструктурная библиотека.

Если в core/ мелькнул такой импорт — файл лежит не там, либо нарушена граница. import-linter ловит это автоматически (R-HEX-TEST-1).

Структура core/

R-HEX-CORE-2: типичная раскладка.

src/orders/
  core/
    order/                           # Bounded Context «Orders»
      aggregate/
        order.py                     # Aggregate Root
      entity/
        order_item.py                # Entity
      value_object/
        money.py                     # Value Object
        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            # Protocol — outbound-порт
          notification_port.py       # Protocol — outbound-порт
    usecase/
      command/
        create_order_command.py      # dataclass-команда + Handler
        confirm_order_command.py
      query/
        get_order_query.py           # dataclass-запрос + Handler
    service/                         # shared domain-логика (по необходимости)

Обрати внимание:

core/order/ — bounded context группирует всё вокруг одной предметной области. Один сервис может иметь 1–3 BC.
port/out/ — Protocol-интерфейсы. Это «что core нужно от внешнего мира»: репозитории, клиенты внешних систем, event publishers.
usecase/ — Command/Query dataclass + Handler пары. Команды меняют состояние агрегата, query возвращают read-проекции.
service/ — shared domain-логика, которая не помещается в один агрегат. Используется редко; чаще — domain-метод или domain event.

Rich domain — методы внутри агрегата

R-HEX-CORE-4: бизнес-логика живёт внутри агрегата, не в *Service-классах.

# 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
        self._events.append(...)

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

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)

Что не так с анемичной моделью:

Инварианты разъезжаются. Логика confirm() дублируется в роутере, в Kafka-consumer'е, в CLI — одна из копий отстанет.
Unit-тест на агрегат невозможен. Order без логики тестируется только через Service с поднятой инфраструктурой.
Lifecycle нечитаем. «Найди все места, где меняется status» вместо одного метода confirm().

Rich domain в DDD — в DDD Tactical Style Guide — R-AGG-* про агрегаты, R-VO-* про value objects.

Outbound-порт как Protocol

R-HEX-PORT-1 / R-HEX-PORT-2: порт объявляется в core/<bc>/port/out/ и оперирует только domain-типами.

# 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 — структурная типизация: out-adapter реализует его без явного наследования, достаточно совпадения сигнатур. Это упрощает подмену в тестах.

Port-исключения объявлены в core/ (R-HEX-PORT-3):

# 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

Подклассы с деталями инфраструктуры (SberPaymentError, PersistenceError) — в out-adapter'ах. Handler ловит базовый OrderNotFoundError, не специфический.

DI без фреймворк-аннотаций

R-HEX-CORE-3: в Python нет Spring-сканирования — core/-классы не нуждаются ни в каких декораторах @injectable, @service. Чистые Python-классы с конструктором через __init__.

Wiring — в app/container.py (R-HEX-BOOT-1):

# 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. Это позволяет в тестах подменить OrderRepository на in-memory реализацию без поднятия базы:

# tests/unit/core/test_confirm_order.py
import pytest
from orders.core.order.aggregate.order import Order, OrderStatus
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

Тест на core/ — молниеносный, без PostgreSQL, без FastAPI.

Контракт import-linter

R-HEX-CORE-X1 / R-HEX-MOD-X1: самодисциплина в Python не работает — нужен автомат.

# 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/ появится from sqlalchemy import ... — CI упадёт с читаемым сообщением об ошибке ещё до code review. Это R-HEX-TEST-2: обязательный required check, PR не мерджится при падении.

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

Антипаттерн	Правило	Что взамен
`from fastapi import HTTPException` в `core/`	`R-HEX-CORE-X1`	Доменное исключение `OrderNotFoundError` в `core/exception/`; маппинг в HTTPException — в роутере
`from sqlalchemy.orm import DeclarativeBase` в `core/`	`R-HEX-CORE-X2`	SQLAlchemy-модель в `adapters/out/persistence/`; маппинг `Order` ↔ ORM-модель — в `order_mapper.py`
`class Order` без методов, только поля + `setter`; вся логика в `OrderService`	`R-HEX-CORE-X3`	Rich domain: `order.confirm()`, `order.cancel(reason)` в агрегате
SQLAlchemy-модель `OrderModel` используется как доменный тип в порте	`R-HEX-CORE-X4`	Port оперирует `Order` (dataclass); маппинг `OrderModel ↔ Order` в `order_mapper.py` persistence-адаптера
Pydantic `CreateOrderRequest` объявлен в `core/order/command.py`	`R-HEX-CORE-X5`	Pydantic-схема в `adapters/in/http/user/schemas.py`; в `core/` — только `CreateOrderCommand` (dataclass)
`Protocol` порта объявлен в `adapters/out/persistence/`	`R-HEX-PORT-X1`	Порт — контракт core-к-инфраструктуре, живёт в `core/<bc>/port/out/`

Куда дальше

Adapters in — как FastAPI-роутер маппит Pydantic DTO → UseCase command, правила in-adapter.
Adapters out — как out-adapter реализует port-Protocol, маппер domain ↔ DTO внешней системы.
Ports — как объявить outbound-порт в core/<bc>/port/out/, что попадает в сигнатуру.
Структура модулей — полная раскладка пакетов и контракт import-linter.
Bootstrap / composition root — как app/container.py и create_app() связывают роутеры, Dispatcher и адаптеры.
Архитектурные тесты — lint-imports в CI как единственный надёжный guard границ.
Когда переходить на Hexagonal — признаки готовности сервиса к Уровню 3.