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

Когда проект только начинается, все файлы лежат в одной папке и всё понятно. Когда он вырастает, возникает вопрос: куда что класть? Можно организовать файлы по типу (models/, services/, routes/) — но тогда весь код одной фичи размазан по разным папкам. Hexagonal Architecture предлагает другой ответ: раскладывать по смыслу и направлению зависимостей.

Три зоны: core, adapters, app

В Hexagonal Architecture весь код делится на три зоны с чёткими правилами.

src/<service>/
├── core/          # бизнес-логика, ничего не знает о FastAPI и базах
├── adapters/      # связь с внешним миром: HTTP, БД, очереди
└── app/           # собирает всё вместе, точка входа

Стрелка зависимостей строго односторонняя:

app → adapters → core

core/ не знает о adapters/. adapters/ не знают друг о друге. app/ знает обо всём и соединяет части.

Почему Python требует import-linter

В Java нарушение границ между слоями не компилируется: если core/build.gradle.kts не объявил зависимость от Spring, компилятор откажется собирать код с import org.springframework. В Python такой защиты нет. Один случайный import sqlalchemy внутри core/ пройдёт незамеченным через IDE, тесты и ревью.

Именно поэтому в Python-проекте на Hexagonal нужен import-linter. Он проверяет границы между слоями во время сборки. Без него архитектура существует только на словах.

Контракт объявляется в pyproject.toml:

[tool.importlinter]
root_package = "service"

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

Слои перечислены от «верхнего» к «нижнему»: app может импортировать adapters и core; adapters — только core; core — никого из этого списка. Нарушение выглядит так:

ImportContractViolation: Module 'service.core.order.aggregate.order'
imports 'service.adapters.out.persistence.models' — violates layers contract.

lint-imports запускается в CI как обязательная проверка. Без зелёного lint-imports ветка не вливается.

Полная структура проекта

Типичный сервис на FastAPI и SQLAlchemy:

src/<service>/
├── core/
│   └── order/
│       ├── aggregate/        # Order, OrderLine (богатая доменная модель)
│       ├── value_object/     # Money, CustomerId
│       ├── event/            # OrderConfirmed, OrderCancelled
│       ├── port/
│       │   └── out/          # OrderRepository, PaymentPort (Protocol)
│       └── usecase/          # ConfirmOrderCommand, ConfirmOrderHandler
├── adapters/
│   ├── in/
│   │   └── http/
│   │       ├── user/         # роутеры для клиентских запросов
│   │       └── admin/        # роутеры для внутренних операций
│   └── out/
│       ├── persistence/      # SQLAlchemy-репозитории
│       ├── sber/             # httpx-клиент к внешнему API
│       └── kafka/            # Kafka producers
└── app/                      # composition root
    ├── main.py               # create_app(), lifespan
    ├── container.py          # DI-wiring портов на адаптеры
    └── settings.py           # pydantic-settings

Минимальный набор для нового сервиса — core/, adapters/out/persistence/, adapters/in/http/user/, app/. Дальше пакеты добавляются по мере появления новых внешних систем и точек входа.

core/ — зона без инфраструктуры

core/ содержит бизнес-логику: агрегаты, value objects, события, порты и хендлеры. Здесь нет ни import fastapi, ни import sqlalchemy, ни import pydantic. Только стандартная библиотека и собственные доменные типы.

Это даёт три практических преимущества:

  • Быстрые тесты. Агрегаты и хендлеры тестируются без поднятия FastAPI, без Testcontainers, без реальной базы. Это обычный Python-метод — его можно вызвать напрямую.
  • Переносимость. Тот же core/ легко завернуть в CLI-скрипт, задачу Celery или облачную функцию, не меняя доменный код.
  • Понятная граница. Новый разработчик открывает pyproject.toml, видит контракт и понимает правила без отдельного документа.

Порты (интерфейсы для внешних систем) объявляются как Protocol:

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

class OrderRepository(Protocol):
    async def get(self, order_id: str) -> Order: ...
    async def save(self, order: Order) -> None: ...

Хендлер принимает порты через конструктор, ничего не зная о реализациях:

# core/order/usecase/confirm_order.py
class ConfirmOrderHandler:
    def __init__(
        self,
        orders: OrderRepository,
        payment: PaymentPort,
    ) -> None:
        self._orders = orders
        self._payment = payment

    async def handle(self, cmd: ConfirmOrderCommand) -> None:
        order = await self._orders.get(cmd.order_id)
        order.confirm()
        await self._payment.charge(order.id, order.total)
        await self._orders.save(order)

adapters/out/ — один пакет на каждую внешнюю систему

Каждая внешняя система получает свой пакет в adapters/out/:

adapters/out/
├── persistence/      # PostgreSQL через SQLAlchemy (async)
├── sber/             # внешнее платёжное API: httpx + DTO + маппер
├── kafka/            # aiokafka producers
└── s3/               # объектное хранилище

Каждый пакет реализует один или несколько протоколов из core/. Пакеты не зависят друг от друга. Если нужно вызвать платёжный адаптер после записи в базу, это делает хендлер в core/ — он получает оба порта и координирует их.

Частая ошибка — собрать всё в один файл adapters/out/mega_adapter.py. Тогда любое изменение в одной системе затрагивает код других, а отдельные настройки надёжности (таймауты, повторные попытки) приходится перемешивать.

adapters/in/ — один пакет на каждый тип входа

Входящие адаптеры тоже разделяются:

adapters/in/
├── http/
│   ├── user/         # APIRouter для клиентских запросов
│   └── admin/        # APIRouter для внутренних операций
└── kafka/            # consumer-хендлеры

Зачем разделять user/ и admin/:

  • Разный контроль доступа. user/-роутер проверяет JWT клиента; admin/-роутер — токен с отдельным audience или mTLS. Их нельзя перепутать, потому что они монтируются отдельно с разными зависимостями.
  • Разные схемы запросов. CreateOrderRequest клиента и AdminCreateOrderRequest — разные Pydantic-модели. Это исключает случайное попадание внутренних полей в клиентский контракт.
# app/main.py
from service.adapters.in_.http.user import router as user_router
from service.adapters.in_.http.admin import router as admin_router

def create_app() -> FastAPI:
    app = FastAPI(lifespan=lifespan)
    app.include_router(user_router, prefix="/api/v1", dependencies=[Depends(verify_user_jwt)])
    app.include_router(admin_router, prefix="/internal", dependencies=[Depends(verify_admin_token)])
    return app

app/ — composition root

app/ — это точка сборки. Она знает обо всех адаптерах и core/, соединяет их вместе и запускает приложение. Никакой бизнес-логики здесь нет.

container.py связывает порты с реализациями:

# app/container.py
from service.adapters.out.persistence.order_repository import SqlOrderRepository
from service.adapters.out.sber.payment_adapter import SberPaymentAdapter
from service.core.order.usecase.confirm_order import ConfirmOrderHandler

class Container:
    def __init__(self, settings: Settings, session_factory) -> None:
        self.order_repo = SqlOrderRepository(session_factory)
        self.payment = SberPaymentAdapter(
            base_url=settings.sber_url,
            api_key=settings.sber_api_key,
        )
        self.confirm_order = ConfirmOrderHandler(
            orders=self.order_repo,
            payment=self.payment,
        )

lifespan управляет ресурсами — открывает соединения при старте и закрывает при остановке:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.connect() as conn:
        await conn.execute(text("SELECT 1"))   # проверка при старте
    yield
    await engine.dispose()

Частые ошибки

Нет import-linter-контракта. Папки core/, adapters/ есть, но никто не проверяет, что границы соблюдаются. Через месяц в core/ появится import sqlalchemy. Решение: добавить layered-контракт в pyproject.toml и lint-imports в CI.

ORM-модель как доменный тип. SQLAlchemy-модель оказывается в core/ — и архитектура рассыпается. Решение: в core/ живёт чистый Python-класс агрегата; маппинг ORM ↔ домен делается в adapters/out/persistence/<x>_mapper.py.

Wiring в адаптере. adapters/out/sber/__init__.py сам создаёт клиент и регистрирует зависимости. Решение: адаптеры — чистые Python-классы без побочных эффектов при импорте; всё wiring — только в app/container.py.

User и admin роутеры в одном пакете. Тогда нельзя применить разные правила проверки токенов без запутанных условий. Решение: сразу создать adapters/in/http/user/ и adapters/in/http/admin/ с отдельными зависимостями.

Коротко

  • Три зоны: core/ (бизнес-логика), adapters/ (связь с внешним миром), app/ (сборка). Стрелка зависимостей: app → adapters → core.
  • core/ не импортирует FastAPI, SQLAlchemy, Pydantic — только стандартную библиотеку и собственные типы.
  • В Python нет compile-time защиты границ. import-linter с layered-контрактом — единственный инструмент, который держит архитектуру честной. Запускается в CI как обязательная проверка.
  • На каждую внешнюю систему — отдельный пакет в adapters/out/: persistence/, sber/, kafka/. Пакеты не зависят друг от друга.
  • Разные точки входа — разные пакеты в adapters/in/: http/user/ и http/admin/ монтируются с разными правилами доступа.
  • app/container.py — единственное место, где порты связываются с реализациями. Никакого wiring в адаптерах.

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