Структура пакетов в Hexagonal Architecture (Python)

Пакетная раскладка Hexagonal в Python: core/ без FastAPI/SQLAlchemy, import-linter как compile-time guard, стрелка app → adapters → core и per-system изоляция адаптеров.

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

Опирается на правила: R-HEX-MOD-1 … R-HEX-MOD-5 и R-HEX-MOD-X1 … R-HEX-MOD-X3 → раздел 2. Структура модулей.

Важно знать

В Python нет compile-time изоляции модулей — import-linter с layered-контрактом закрывает эту брешь.

core/ не импортирует FastAPI, SQLAlchemy, Pydantic — только stdlib и доменные типы.

Per-system: adapters/out/sber/, adapters/out/persistence/, adapters/out/kafka/ — отдельный пакет на каждую внешнюю систему.

Per-purpose: adapters/in/http/user/, adapters/in/http/admin/ — отдельные пакеты, отдельный security-профиль.

app/ — composition root: create_app(), контейнер DI-wiring, lifespan, settings. Никакой бизнес-логики.

Стрелка зависимостей строго: app → adapters → core. Адаптеры не знают друг о друге.

import-linter запускается в CI как required check — PR без зелёного lint-imports не мерджится.

В Java compile-time граница держится Gradle-модулями: если core/build.gradle.kts не объявил зависимость от Spring, компилятор не даст импортировать @Service. Python такой гарантии не даёт — один import sqlalchemy в файле внутри core/ пройдёт незамеченным через IDE, тесты и ревью. Именно поэтому import-linter здесь не «nice to have», а единственный механизм, который держит Hexagonal честным. Ниже — раскрытие правил R-HEX-MOD-* в идиомах Python.

Пакетная раскладка

R-HEX-MOD-1 — типичная структура сервиса на FastAPI/SQLAlchemy:

src/<service>/
├── core/
│   └── order/
│       ├── aggregate/        # Order, OrderLine (rich domain)
│       ├── 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-клиент к Sber-API
│       └── kafka/            # Kafka producers
└── app/                      # composition root
    ├── main.py               # create_app(), lifespan
    ├── container.py          # DI-wiring портов на адаптеры
    └── settings.py           # pydantic-settings

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

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

R-HEX-MOD-2: core/ зависит только от stdlib и доменных протоколов. В pyproject.toml на уровне пакета core/ не объявляется ни fastapi, ни sqlalchemy, ни pydantic — только, если нужно, typing-extensions или специализированные DDD-утилиты.

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 — никого из этого списка. Нарушение — lint-imports падает с явным трейсом до виновного импорта.

Что это даёт на практике:

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

Per-system out-адаптеры

R-HEX-MOD-3: на каждую внешнюю систему — отдельный пакет в adapters/out/.

adapters/out/
├── persistence/      # PostgreSQL через SQLAlchemy (async)
├── sber/             # Sber-API: httpx-клиент + DTO + маппер
├── kafka/            # aiokafka producers
└── s3/               # boto3 / aiobotocore для объектного хранилища

Каждый из этих пакетов реализует один или несколько Protocol-портов из core/<bc>/port/out/. Они не зависят друг от друга — если SberPaymentAdapter нужно вызвать после записи в PG, это координирует ConfirmOrderHandler в core/, инжектируя оба порта:

# 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)

Resilience-конфиг (таймауты, circuit breaker через tenacity или circuitbreaker) настраивается отдельно для каждого адаптера — это правило R-RES-ISO-1.

Per-purpose in-адаптеры

R-HEX-MOD-4: каждый тип входа — свой пакет.

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

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

Разный security-профиль. user/-роутер верифицирует JWT клиента; admin/-роутер — токен с отдельным audience или mTLS. В app/create_app() они монтируются раздельно, с разными dependencies:

# 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

Разные Pydantic-модели. CreateOrderRequest клиента и AdminCreateOrderRequest (с override-полями) — разные схемы, не общая. Это исключает случайное протекание admin-полей в user-контракт.

app/ — composition root

R-HEX-MOD-5: app/ собирает всё. Зависит от core/ и всех адаптеров; от app/ не зависит никто.

# 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,
        )

create_app() принимает Container, регистрирует роутеры и возвращает сконфигурированный FastAPI-экземпляр. Никакой бизнес-логики, никаких if-веток по домену.

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

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(lambda c: None)   # прогрев пула
    yield
    await engine.dispose()

Стрелка зависимостей

app → adapters → core

app/ импортирует адаптеры и core/.
Каждый адаптер импортирует только core/.
core/ не импортирует ничего за пределами stdlib.
Адаптеры не импортируют друг друга.

Когда кажется, что core/ нужен репозиторий — это сигнал, что в core/ оказалось то, чему там не место (ORM-модель, generated DTO). Когда кажется, что sber/-адаптер нужен persistence/ — это сигнал вынести координацию в handler в core/.

Нарушение сразу видно в lint-imports:

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

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

Антипаттерн	Правило	Что взамен
Единый пакет без import-linter-контракта — «папки `core/`, `adapters/` есть, но проверки нет»	`R-HEX-MOD-X1`	Объявить layered-контракт в `pyproject.toml`, добавить `lint-imports` в CI
`core/<bc>/` импортирует `adapters/out/persistence/models.py`	`R-HEX-MOD-X2`	Объявить `Protocol`-порт в `core/<bc>/port/out/`; маппинг ORM-модели ↔ domain — в `adapters/out/persistence/<x>_mapper.py`
User- и admin-роутеры в одном пакете `adapters/in/http/`	`R-HEX-MOD-X3`	Разделить на `adapters/in/http/user/` и `adapters/in/http/admin/` с отдельными `dependencies` в `include_router`
FastAPI-импорт в `core/`	`R-HEX-CORE-X1`	FastAPI — деталь in-adapter; `core/` видит только stdlib
SQLAlchemy ORM-модель как доменный тип в `core/`	`R-HEX-CORE-X4`	Отдельный domain-aggregate; маппинг в `adapters/out/persistence/<x>_mapper.py`
Один `adapters/out/mega_adapter.py` реализует порты нескольких доменов	`R-HEX-AOUT-X3`	Per-system пакеты: `adapters/out/sber/`, `adapters/out/kafka/` — каждый со своим маппером и resilience-конфигом
Wiring адаптеров на порты в `adapters/out/<system>/__init__.py`	`R-HEX-BOOT-X2`	DI-wiring только в `app/container.py`; адаптеры — pure Python-классы без side-effects при импорте

Куда дальше

Core слой — внутреннее устройство core/: агрегаты, value objects, порты.
Ports — как объявить Protocol-порт и почему не ABC.
Входящие адаптеры — FastAPI-роутеры, Pydantic-DTO, маппинг в UseCase command.
Исходящие адаптеры — SQLAlchemy, httpx, реализация порт-Protocol.
Bootstrap / composition root — create_app(), Container, lifespan.
Архитектурные тесты — import-linter: контракты, forbidden, independence.
Когда переходить на Hexagonal — признаки, что пора, и признаки, что рано.