Bootstrap / Composition Root в Hexagonal (Python)

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

Когда core/ и адаптеры готовы, их ещё нужно соединить: сказать приложению, какой репозиторий подключить к какому порту, где взять настройки, как запустить сервер и как его правильно остановить. Весь этот «монтажный» код живёт в одном месте — app/. Разберём, почему это устроено именно так и как это выглядит на практике.

Зачем вообще нужен composition root

Представьте, что в каждом файле кода написано db_url = os.getenv("DB_URL") — и так в десяти местах. Или что create_async_engine(...) создаётся при импорте модуля, а не при старте приложения. В такой ситуации тесты сложно изолировать, настройки расползлись по всему коду, и нигде не видно полной картины того, как сервис собран.

Composition root — это единственное место в приложении, где собираются все зависимости вместе. Он знает про core/, про все адаптеры, про настройки. Зато всё остальное — core/, адаптеры, роутеры — про app/ не знают ничего.

В гексагональной архитектуре на Python этим местом является папка app/.

Что лежит в app/

Структура сервиса выглядит так:

src/<service>/
  app/
    __init__.py
    main.py          # create_app() + точка запуска
    container.py     # DI-wiring: Handler'ы, репозитории, Dispatcher
    lifespan.py      # @asynccontextmanager: engine, клиенты, пулы
    settings.py      # pydantic-settings BaseSettings
  core/
    order/
      aggregate.py
      port/out/
        order_repository.py     # Protocol
        payment_port.py         # Protocol
      usecase/
        create_order.py         # UseCase + Handler
  adapters/
    in/http/
      order_router.py
    out/
      persistence/
        sqlalchemy_order_repository.py
      sber/
        sber_payment_adapter.py
Dockerfile
docker-compose.yml
pyproject.toml

В app/ нет роутеров и нет доменных классов — только склейка. Роутер живёт в adapters/in/http/, бизнес-логика — в core/.

Settings: единое место для конфигурации

Частая проблема — настройки «россыпью»: os.getenv("DB_URL") в одном файле, os.getenv("API_KEY") в другом. Если переменная не выставлена, узнаёшь об этом в самый неподходящий момент — когда уже обслуживается запрос.

Решение: один Settings-класс через pydantic-settings. Все переменные объявлены явно, типизированы, и валидируются сразу при запуске — приложение не стартует, пока не выставлены все обязательные переменные.

# app/settings.py
from pydantic import PostgresDsn, AnyHttpUrl
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Literal


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

    app_env: Literal["local", "integration-test", "production"] = "local"

    db_url: PostgresDsn
    db_pool_size: int = 10

    sber_api_url: AnyHttpUrl
    sber_api_key: str

    auth_disabled: bool = False
    jwks_url: AnyHttpUrl | None = None

db_url и sber_api_key объявлены без значения по умолчанию — это обязательные поля. Если переменная окружения не выставлена, pydantic упадёт на старте с внятным сообщением, а не посреди обработки запроса.

Профиль среды (local, integration-test, production) задаётся только через переменную окружения APP_ENV — не кодом, не условиями внутри приложения.

App factory: фабрика вместо глобала

Ещё одна частая ошибка — создавать приложение прямо на уровне модуля:

# Так делать не надо
app = FastAPI()
app.include_router(order_router)

Проблема: при запуске тестов этот код выполняется при импорте, со всеми глобальными зависимостями. Запустить два теста с разными настройками невозможно.

Правильный подход — фабричная функция create_app(settings):

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from .settings import Settings
from .container import Container
from .lifespan import build_lifespan


def create_app(settings: Settings | None = None) -> FastAPI:
    if settings is None:
        settings = Settings()

    container = Container(settings=settings)
    lifespan = build_lifespan(container)

    app = FastAPI(
        title="order-service",
        lifespan=lifespan,
    )

    _register_routers(app, container)
    _register_exception_handlers(app)

    return app


def _register_routers(app: FastAPI, container: Container) -> None:
    from service.adapters.in_.http.order_router import make_router
    app.include_router(make_router(container))


def _register_exception_handlers(app: FastAPI) -> None:
    from service.adapters.in_.http.exception_handlers import register
    register(app)


app = create_app()

Тест вызывает create_app(FakeSettings(...)) и получает полностью изолированный экземпляр приложения. make_router(container) — роутер принимает зависимости явно через параметр, а не через глобальный импорт.

Lifespan: управление ресурсами

Подключение к базе данных, пул соединений, HTTP-клиент к внешнему сервису — эти ресурсы нужно открывать при старте и закрывать при остановке. Если создать их при импорте модуля, теряется контроль над порядком инициализации и освобождения.

FastAPI решает это через lifespan — специальный контекстный менеджер:

# app/lifespan.py
from contextlib import asynccontextmanager
from collections.abc import AsyncGenerator
from fastapi import FastAPI
import httpx
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from .container import Container


def build_lifespan(container: Container):
    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
        settings = container.settings

        engine = create_async_engine(
            str(settings.db_url),
            pool_size=settings.db_pool_size,
            echo=settings.app_env == "local",
        )
        session_factory = async_sessionmaker(engine, expire_on_commit=False)
        container.session_factory.override(session_factory)

        async with httpx.AsyncClient(base_url=str(settings.sber_api_url)) as sber_client:
            container.sber_http_client.override(sber_client)
            yield

        await engine.dispose()

    return lifespan

Всё, что до yield — запуск. Всё после yield — остановка. При получении SIGTERM uvicorn/gunicorn завершит обработку текущих запросов и выполнит блок после yield. engine.dispose() закрывает пул соединений с базой данных.

DI-контейнер: wiring core и адаптеров

container.py — это карта зависимостей: какой адаптер подключён к какому порту. Пример с библиотекой dependency-injector:

# app/container.py
from dependency_injector import containers, providers
from .settings import Settings

from service.core.order.port.out.order_repository import OrderRepository
from service.core.order.port.out.payment_port import PaymentPort
from service.core.order.usecase.create_order import CreateOrderHandler
from service.core.order.usecase.get_order import GetOrderHandler
from service.adapters.out.persistence.sqlalchemy_order_repository import SqlAlchemyOrderRepository
from service.adapters.out.sber.sber_payment_adapter import SberPaymentAdapter
from service.core.shared.dispatcher import Dispatcher


class Container(containers.DeclarativeContainer):
    settings: providers.Object[Settings] = providers.Object(Settings())

    session_factory = providers.Object(None)
    sber_http_client = providers.Object(None)

    order_repository: providers.Provider[OrderRepository] = providers.Factory(
        SqlAlchemyOrderRepository,
        session_factory=session_factory,
    )

    payment_port: providers.Provider[PaymentPort] = providers.Factory(
        SberPaymentAdapter,
        http_client=sber_http_client,
        api_key=providers.Callable(lambda s: s.sber_api_key, settings),
    )

    create_order_handler: providers.Provider[CreateOrderHandler] = providers.Factory(
        CreateOrderHandler,
        order_repository=order_repository,
        payment_port=payment_port,
    )

    get_order_handler: providers.Provider[GetOrderHandler] = providers.Factory(
        GetOrderHandler,
        order_repository=order_repository,
    )

    dispatcher: providers.Provider[Dispatcher] = providers.Singleton(
        Dispatcher,
        handlers={
            "CreateOrder": create_order_handler,
            "GetOrder": get_order_handler,
        },
    )

SqlAlchemyOrderRepository реализует Protocol OrderRepository; SberPaymentAdapter — Protocol PaymentPort. core/ не знает про конкретные классы адаптеров — только про Protocol-интерфейсы.

Недетерминированные источники (clock, id_generator) тоже оборачиваются в провайдеры:

    clock: providers.Provider = providers.Factory(
        lambda: __import__("datetime").datetime.now,
    )
    id_generator: providers.Provider = providers.Factory(
        lambda: __import__("uuid").uuid4,
    )

В тесте container.clock.override(lambda: fixed_datetime) даёт детерминированный результат без патчинга глобальных функций.

Как роутер получает зависимости

Роутер не импортирует контейнер напрямую — он получает зависимости через параметр и передаёт их в Depends:

# adapters/in_/http/order_router.py
from fastapi import APIRouter, Depends
from service.core.shared.dispatcher import Dispatcher
from .schemas import CreateOrderRequest, OrderResponse
from .order_request_mapper import to_command, to_response


def make_router(container) -> APIRouter:
    router = APIRouter(prefix="/orders", tags=["orders"])

    def get_dispatcher() -> Dispatcher:
        return container.dispatcher()

    @router.post("/", response_model=OrderResponse, status_code=201)
    async def create_order(
        body: CreateOrderRequest,
        dispatcher: Dispatcher = Depends(get_dispatcher),
    ) -> OrderResponse:
        command = to_command(body)
        result = await dispatcher.dispatch(command)
        return to_response(result)

    return router

Роутер знает только про Dispatcher — не про SqlAlchemyOrderRepository или SberPaymentAdapter. Детали реализации скрыты за контейнером.

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

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

app/ ──depends──→ core/
app/ ──depends──→ adapters/in/http/
app/ ──depends──→ adapters/out/persistence/
app/ ──depends──→ adapters/out/sber/

adapters/in/http/ ──depends──→ core/
adapters/out/* ──depends──→ core/

core/ ──зависит ТОЛЬКО от stdlib──→ ∅

core/ не импортирует adapters/ или app/. adapters/in/* не импортирует adapters/out/*. Это проверяется автоматически через import-linter в CI:

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

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

Если кто-то случайно добавит from service.app.container import container в core/, CI упадёт с понятным сообщением.

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

Роутер или Handler в app/main.py или app/container.py. Роутер принадлежит adapters/in/http/, handler — core/<bc>/usecase/. В app/ только сборка, не логика.

engine = create_async_engine(...) в глобале модуля. Engine должен создаваться в lifespan и передаваться через container override — иначе он создаётся при импорте, до того как выставлены настройки, и не закрывается при остановке.

os.getenv("DB_URL") в нескольких местах кода. Заведите один Settings-объект с BaseSettings и передавайте его туда, где нужно.

from service.app.container import container в core/. Стрелка зависимостей идёт только внутрь: app → adapters → core. core/ не знает про app/.

Бизнес-логика в container.py (if settings.env == "local" — другой handler). Профильное поведение выносят в порт с разными реализациями.

Base.metadata.create_all() в lifespan для продакшна. Миграции управляются через Alembic — в CI и деплое, не в коде приложения.

Локальный запуск

docker compose up -d postgres
alembic upgrade head
uvicorn service.app.main:app --reload

В продакшне рекомендуется gunicorn -k uvicorn.workers.UvicornWorker — он обрабатывает SIGTERM и передаёт управление lifespan для корректной остановки.

Коротко

• app/ — composition root: единственное место, где собираются все зависимости. Никто не импортирует app/ — только app/ импортирует всё остальное.
• Конфигурация — один Settings-класс на BaseSettings. Обязательные поля без default: приложение падает на старте, если переменная не выставлена.
• create_app(settings) — фабрика, а не модуль-уровень глобал. Тесты вызывают её с тестовыми настройками и получают изолированный экземпляр.
• Ресурсы (engine, HTTP-клиенты) — в lifespan. До yield — открываем, после yield — закрываем. Graceful shutdown происходит автоматически.
• container.py — карта зависимостей: какой адаптер реализует какой порт. Роутеры и handlers не знают про конкретные реализации.
• Архитектурный контракт (core/ → ∅, adapters/ → core/, app/ → всё) проверяется import-linter в CI.

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

• Структура пакетов — пакетная раскладка core/adapters/app и import-linter-контракт
• Core слой — что живёт в core/ и почему туда нельзя FastAPI/SQLAlchemy
• Ports — Protocol как outbound-порт, port-исключения, domain-типы в сигнатурах
• Adapters in — роутеры, маппер request-DTO в команду, связь через Dispatcher
• Adapters out — реализация Protocol-портов, domain ↔ DTO маппинг в адаптере
• Архитектурные тесты — import-linter: layers + forbidden + independence контракты в CI