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

Когда 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 dependency_injector import providers
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()
    container.settings.override(providers.Object(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.Provider[Settings] = providers.Singleton(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/ adapters/ core/

Стрелка «зависит от» идёт только внутрь: app знает всё, адаптеры знают core, core никого; обратный импорт ловит import-linter в CI.

Кто зависитОт кого
app/core/, adapters/in_/http/, adapters/out/persistence/, adapters/out/sber/
adapters/in_/http/core/
adapters/out/*core/
core/только стандартная библиотека

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/ только сборка, не логика.

# так делать не нужно: маршрут и правило домена в точке сборки
app = create_app(Settings())


@app.post("/orders")
async def create_order(req: CreateOrderRequest):
    if req.total > 100_000:                   # бизнес-правило в app/main.py
        raise HTTPException(status_code=400)
    ...

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

# так делать не нужно
engine = create_async_engine(os.environ["DB_URL"])   # создаётся при импорте модуля


# правильно: ресурс живёт в lifespan и попадает в контейнер подменой
@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine(settings.database_url)
    container.engine.override(providers.Object(engine))
    yield
    await engine.dispose()

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

В продакшне uvicorn запускают напрямую с явным --timeout-graceful-shutdown: он сам обрабатывает SIGTERM, дожидается начатых запросов и выполняет блок после yield. Если нужен gunicorn, класс воркера берут из пакета uvicorn-worker (модуль uvicorn.workers объявлен устаревшим), а ожидание задаёт --graceful-timeout gunicorn. Подробно порядок остановки разобран в статье про конфигурацию uvicorn и lifespan.

Дополнительно: при первом чтении можно пропустить

Глубже: откуда берутся значения в продерасширенное

db_url: PostgresDsn в Settings это объявление, и вопрос «откуда приходит значение» решается не в приложении, а вокруг него. У pydantic-settings источники читаются в фиксированном порядке приоритета: аргументы конструктора, переменные окружения, файл .env, файлы из secrets_dir, значение по умолчанию. То есть переменная окружения всегда бьёт .env, и локальный файл не мешает контейнеру.

В Kubernetes переменные окружения приходят из ConfigMap через envFrom, а секреты монтируют файлами: SettingsConfigDict(secrets_dir="/run/secrets") читает db_password из файла с таким именем, и пароль не светится в kubectl describe pod среди переменных. Сами значения в Secret кладёт оператор внешнего хранилища, не человек с kubectl, об этом статья про персональные данные и секреты.

Профиль среды это одна переменная APP_ENV, а не три файла настроек в репозитории: что отличается между средами, то приходит снаружи, что одинаково, то в коде по умолчанию. Файл .env лежит в .gitignore и нужен только разработчику.

Проверка, что значения доехали, одна: приложение падает на старте с перечнем недостающих полей, потому что у обязательных полей нет умолчаний. Это лучший момент узнать о пропущенной переменной: до первого запроса.

Глубже: тесты в точке сборкирасширенное

Пакет, который «только клей», оказывается главным тестовым модулем, и это стоит проговорить, потому что удивляет.

Тест подъёма приложения. Один тест, который вызывает create_app(test_settings) и входит в with TestClient(app):, запуская lifespan. Он ловит ровно те ошибки, которые не видят юнит-тесты: провайдер контейнера ссылается на несуществующий аргумент, адаптеру не передали клиента, две зависимости образовали цикл, Settings требует поле, которого нет в тестовом профиле. Падает он за секунду и спасает от выката, который умирает на старте.

# tests/app/test_startup.py
def test_app_starts_with_test_settings():
    app = create_app(Settings(_env_file="tests/.env.test"))
    with TestClient(app) as client:          # входит в lifespan: контейнер собран, ресурсы открыты
        assert client.get("/health").status_code == 200

Тест маршрутов. Пройти по app.routes и проверить, что каждый роутер подключён с нужным префиксом и у операций вне белого списка есть зависимость безопасности; как это делают через app.openapi(), показано в статье про контракт OpenAPI.

Тест контейнера с подменами. container.payment_port.override(FakePayment()) и сценарий через HTTP от роутера до поддельного адаптера: он проверяет проводку, а не бизнес-правила, которые уже покрыты тестами ядра.

# tests/app/test_wiring.py
def test_payment_goes_through_adapter(app, container):
    fake = FakePayment()
    with container.payment_port.override(fake):
        with TestClient(app) as client:
            response = client.post("/orders/42/pay", json={"amount": "100.00"})
    assert response.status_code == 202
    assert fake.registered == [OrderId("42")]

Чего здесь нет: тестов доменных правил через HTTP. Они живут в core/, запускаются без приложения и не должны зависеть от сборки.

Коротко

  • 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.
  • Источники настроек читаются по приоритету: аргументы, окружение, .env, secrets_dir, умолчание; секреты приходят файлами из Secret; в точке сборки живут тест подъёма через TestClient с lifespan, тест маршрутов и тест проводки с подменой портов.

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

  • Структура пакетов — пакетная раскладка 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