Опирается на правила: R-HEX-WHEN-1 … R-HEX-WHEN-3 и R-HEX-WHEN-X1 … R-HEX-WHEN-X2 из Hexagonal Style Guide → раздел 1. Когда переходить на Hexagonal.

Важно знать

  • Hexagonal — часть Уровня 3 в UCP. На Уровне 1–2 — overkill, ceremony без выгоды.
  • В Python нет compile-time изоляции модулей: import-linter — единственный enforcement границ, аналог ArchUnit в Java.
  • Признаки «пора»: 2+ внешних системы, богатый домен с инвариантами, 3+ точки входа, команда от 3 человек, тесты требуют поднять всё приложение.
  • Признаки «рано»: один сервис с одной PG, 1–2 разработчика, < 10K строк, форма бизнес-логики ещё не устоялась.
  • Cargo-cult запрещён (R-HEX-WHEN-X1): сервис из 3 эндпоинтов с полной hex-раскладкой — ceremony без выгоды.
  • Частичный Hexagonal — антипаттерн (R-HEX-WHEN-X2): core/ есть, но роутеры мешают бизнес-логику с HTTP — либо полная схема, либо никакой.
  • Решение принимается на сервис, не на команду: рядом могут жить сервис Уровня 1 и сервис Уровня 3.
  • Отсутствие import-linter-контракта = нет Hexagonal, даже если папки называются core/ и adapters/.

Hexagonal Architecture на Python даёт строгое разделение: core/ без FastAPI, SQLAlchemy и Pydantic, порты-Protocol в доменном слое, адаптеры — только маппинг и инфраструктура, import-linter в CI держит границы. Но за это приходится платить: дополнительные пакеты, mappers между слоями, явный wiring в app/. Переходить стоит ровно тогда, когда выгода измерима.

Hexagonal — часть Уровня 3

R-HEX-WHEN-1: Hexagonal в UCP — часть Уровня 3 (DDD + ports/adapters + import-linter). На Уровнях 1–2 — избыточен.

Шкала зрелости применительно к Python-сервисам:

  • Уровень 1 — плоский app/ с FastAPI-роутерами, простой CRUD, одна PG. UseCase + Handler в одном пакете, без явного hex-разделения. Сервисы на старте или простые внутренние API.
  • Уровень 2 — UseCase + rich domain (агрегаты с инвариантами, order.confirm()), но adapters/ и core/ живут рядом без import-linter-контракта. Бизнес-логика выделена, но архитектурных тестов нет.
  • Уровень 3 — DDD + Hexagonal. Пакеты core/adapters/app/, порты как Protocol, import-linter в CI как required check. Это то, про что эта статья.
# Уровень 1 — плоский app/, всё вместе
src/order_service/
  routers/orders.py      # FastAPI-роутер + бизнес-логика + SQLAlchemy — всё здесь
  models.py

# Уровень 3 — полный hex
src/order_service/
  core/order/{aggregate,port,usecase}/
  adapters/in/http/
  adapters/out/persistence/
  adapters/out/sber/
  app/                   # create_app(), container, lifespan, settings

Переносить Уровень 1 в hex — это как добавить circuit breaker к функции из 30 строк: инструмент существует, но не под эту задачу.

Признаки «пора»

R-HEX-WHEN-2: сигналы, при которых Hexagonal даёт измеримый выигрыш.

1. Сервис интегрируется с 2+ внешними системами. Типичный Уровень 3 — PG + Сбербанк API + Kafka + SMS-шлюз. У каждой системы свои DTO, своя логика retry/timeout, свои failure-режимы. Без явных out-adapter-пакетов:

# Уровень 1: всё в одном handler — быстро растёт неуправляемо
class ConfirmOrderHandler:
    async def handle(self, cmd: ConfirmOrderCommand) -> None:
        async with self.session.begin():
            order = await self.session.get(OrderOrmModel, cmd.order_id)
            sber_resp = await self.http.post(
                "https://securepayments.sberbank.ru/payment/rest/register.do",
                json={"amount": order.total_kopecks, "orderNumber": str(order.id)},
            )
            if sber_resp.json()["errorCode"] != "0":
                raise ValueError(sber_resp.json()["errorMessage"])
            order.status = "confirmed"

С явными портами каждая внешняя система заменяется в тестах без поднятия реального HTTP:

# core/order/port/out/payment_port.py
class PaymentPort(Protocol):
    async def register_payment(self, order: Order) -> PaymentRef: ...

# adapters/out/sber/sber_payment_adapter.py
class SberPaymentAdapter:
    async def register_payment(self, order: Order) -> PaymentRef:
        req = SberMapper.to_register_request(order)
        resp = await self._client.post("/payment/rest/register.do", json=req)
        return SberMapper.to_payment_ref(resp.json())

2. Домен богатый: агрегаты с инвариантами. Заказ с подтверждением, проверкой остатков, событиями — order.confirm() проверяет 3 условия и эмитит OrderConfirmed. Это логика, которую нужно тестировать чистым Python без поднятия FastAPI или SQLAlchemy:

# core/order/aggregate.py — pure Python, нет ни FastAPI, ни SQLAlchemy
@dataclass
class Order:
    id: OrderId
    customer_id: CustomerId
    items: list[OrderItem]
    status: OrderStatus
    events: list[DomainEvent] = field(default_factory=list)

    def confirm(self) -> None:
        if self.status != OrderStatus.PENDING:
            raise OrderAlreadyConfirmedError(self.id)
        if not self.items:
            raise EmptyOrderError(self.id)
        self.status = OrderStatus.CONFIRMED
        self.events.append(OrderConfirmed(order_id=self.id))

Тест запускается за миллисекунды, без pytest-anyio и тестовой БД:

def test_confirm_empty_order_raises():
    order = Order(id=OrderId.generate(), customer_id=CustomerId("C-1"), items=[], status=OrderStatus.PENDING)
    with pytest.raises(EmptyOrderError):
        order.confirm()

3. 3+ точки входа с разными security-контекстами. HTTP для клиента + HTTP для оператора + Kafka consumer + scheduled task — каждый entry point имеет свою модель авторизации. Без разделения пакетов security-middleware начинает смешиваться:

adapters/in/http/customer/   # JWT, публичные эндпоинты
adapters/in/http/operator/   # service-to-service mTLS
adapters/in/kafka/           # consumer, нет HTTP-auth
adapters/in/cron/            # scheduled jobs

import-linter с контрактом independence гарантирует, что adapters/in/http/operator не импортирует из adapters/in/http/customer.

4. Тесты требуют поднять всё приложение. Если для проверки бизнес-правила нужен TestClient(app) — потому что ConfirmOrderHandler импортирует SQLAlchemy-модель напрямую — Hexagonal решает это чистым core/ без инфраструктурных зависимостей.

5. Команда от 3 разработчиков. Архитектурные границы становятся социальным инструментом. lint-imports в CI ловит случайный from sqlalchemy.orm import Session в core/order/aggregate.py там, где code review пропустит:

# pyproject.toml
[[tool.importlinter.contracts]]
name = "core-must-not-import-infrastructure"
type = "forbidden"
source_modules = ["order_service.core"]
forbidden_modules = ["fastapi", "sqlalchemy", "pydantic", "httpx"]

Если хотя бы 3 из 5 — переходим на Hexagonal.

Признаки «рано»

R-HEX-WHEN-3: когда переход даст больше overhead, чем пользы.

Один сервис с одной PG, простой CRUD. Product-каталог Сбера — получить список товаров, обновить цену. Внешний мир — только PG. Repository-pattern в плоском app/ достаточно:

# Уровень 1: плоский app/, репозиторий через SQLAlchemy — для CRUD-каталога этого хватает
class ProductRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def find_by_id(self, product_id: ProductId) -> Product | None:
        row = await self._session.get(ProductOrmModel, product_id.value)
        return ProductMapper.to_domain(row) if row else None

Добавить порт-Protocol, отдельный пакет адаптера и import-linter-контракт ради одной PG — ceremony без выгоды.

1–2 разработчика, < 10K строк. Конвенции держатся устно, core/ из трёх файлов не даёт никому повода нарушить границы. Архитектурный тест — overhead там, где проще договориться.

Бизнес-логика активно ищет форму. В стартапе первые 6–12 месяцев «Customer» может превратиться в «Account», «Order» — в «Contract». Hex-структура с её mappers и Protocol-портами тормозит итерации: каждое «давай попробуем иначе» = переписывание трёх mappers и трёх портов. Сначала найти устойчивую форму, потом — Hexagonal.

Нет агрегатов с инвариантами. Если Customer — это dataclass с полями и без методов, а вся «логика» — это update_email(new_email) без проверок, hex-разделение ничего не защищает. Анемичный домен в hex-обёртке — дорогой способ писать процедурный код.

Cargo-cult и частичный Hexagonal — запрет

R-HEX-WHEN-X1: cargo-cult — все сервисы команды причёсаны под hex независимо от сложности. Сервис из 3 эндпоинтов GET /products, POST /products, DELETE /products/{id} в полной hex-раскладке:

src/product_service/
  core/product/{aggregate.py,port/,usecase/}     # 3 файла
  adapters/in/http/product_router.py             # 20 строк
  adapters/out/persistence/product_adapter.py    # 30 строк
  app/container.py                               # wiring 3 объектов

Это ~200 строк дополнительного кода ради полной изоляции домена, который не имеет инвариантов и никогда не заменит PG на другое хранилище. Решение про переход принимается на сервис: рядом с биллингом Уровня 3 живёт каталог Уровня 1.

R-HEX-WHEN-X2: частичный Hexagonal — core/ создан, но роутеры содержат бизнес-логику, или SQLAlchemy-модели используются в core/ напрямую:

# Антипаттерн: "есть core/, но граница не держится"
# adapters/in/http/order_router.py — роутер принимает решения
@router.post("/orders/{order_id}/confirm")
async def confirm_order(order_id: UUID, session: AsyncSession = Depends(get_session)):
    order_row = await session.get(OrderOrmModel, order_id)  # SQLAlchemy в роутере
    if order_row.total > 50_000:                            # бизнес-логика в роутере
        raise HTTPException(status_code=400, detail="Превышен лимит")
    order_row.status = "confirmed"
    await session.commit()

Что в этом плохого:

  • import-linter не ловит нарушения, потому что граница не задана контрактом — нарушения накапливаются незаметно.
  • Тесты на бизнес-логику требуют HTTP: TestClient + тестовая БД вместо order.confirm().
  • Mental overhead: читая сервис, разработчик не знает, где граница — в каждом файле нужно проверять.

Правило: либо полный Hexagonal (все пакеты, import-linter-контракт в CI, mappers), либо никакого. Переходный период допустим только с явным сроком.

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

АнтипаттернПравилоЧто взамен
Hex на сервис из 3 эндпоинтов без сложного доменаR-HEX-WHEN-X1Уровень 1: плоский app/, Repository-pattern
core/ есть, но роутеры содержат бизнес-логикуR-HEX-WHEN-X2Полная схема: core/adapters/app/ + import-linter-контракт
Hex без import-linter-контрактаR-HEX-MOD-X1pyproject.toml с [tool.importlinter], lint-imports в CI
FastAPI-импорт в core/R-HEX-CORE-X1core/ зависит только от stdlib + доменных типов
SQLAlchemy-модель в сигнатуре портаR-HEX-PORT-X2Порт оперирует domain-типами; маппинг — в адаптере
Бизнес-логика в роутере или out-adapterR-HEX-AIN-X1, R-HEX-AOUT-X2Логика — в core/<bc>/usecase/, адаптер только маппит

Куда дальше

  • Структура модулей — как организовать пакеты core/adapters/app/ и настроить import-linter-контракт.
  • Core слой — что попадает в core/, какие зависимости допустимы.
  • Ports — как описывать outbound-порты через Protocol в Python.
  • Adapters in — роутер FastAPI: маппинг Pydantic DTO → UseCase command, вызов Dispatcher.
  • Adapters out — реализация порта-Protocol, маппинг domain ↔ DTO внешней системы.
  • Bootstrap — create_app(), wiring портов на адаптеры, lifespan.
  • Архитектурные тесты — import-linter: контракт layers, forbidden, independence в CI.