Опирается на правила: R-HEX-BOOT-1 … R-HEX-BOOT-3, R-HEX-BOOT-X1 … R-HEX-BOOT-X2 и PYBOOT-5 … PYBOOT-12 из Hexagonal Style Guide → раздел 7. Bootstrap / composition root.

Важно знать

  • app/ — composition root: create_app(), container.py, lifespan, settings, Dockerfile. Роутеры и бизнес-логика — не здесь.
  • app/ зависит от core/ и всех адаптеров. От app/ не зависит никто — это закрывающий узел.
  • Приложение собирается фабрикой create_app(settings) -> FastAPI, не модуль-левел глобалом — чтобы тесты поднимали изолированные инстансы (PYBOOT-5).
  • Ресурсы (engine, httpx-клиенты) — через @asynccontextmanager lifespan, не глобальными синглтонами на импорте (PYBOOT-6).
  • BaseSettings от pydantic-settings валидируется на старте (fail-fast); профиль задаётся только через APP_ENV, не кодом (PYBOOT-2/PYBOOT-3/PYBOOT-4).
  • DI-контейнер (dependency-injector / punq) собирает Handler'ы, репозитории, Dispatcher в app/container.py (PYBOOT-8, R-HEX-BOOT-1).
  • Graceful shutdown реализован через lifespan + обработку SIGTERM; в prod — gunicorn -k uvicorn.workers.UvicornWorker (PYBOOT-12).

app/ — маленький, но принципиальный модуль. core/ и адаптеры — это «кирпичи»; app/ — «дом», который собирает их вместе. Он знает обо всех частях; никто не знает про app/. Это и есть composition root.

Что лежит в app/

R-HEX-BOOT-1:

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

Это раскладка из R-HEX-MOD-1. В app/ нет роутеров и нет доменных классов — только склейка.

Settings: профили через APP_ENV

PYBOOT-2/PYBOOT-3/PYBOOT-4 — конфиг через pydantic-settings. Три среды (local, integration-test, production); переключение — исключительно через env.

# 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 — required без default: если env не выставлен, pydantic упадёт на старте с внятным сообщением, а не в процессе первого запроса. PYBOOT-X1 — os.getenv() россыпью вместо одного Settings-объекта — запрещён.

App factory

PYBOOT-5 — фабрика, не глобал:

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

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

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

PYBOOT-6 — engine, пулы, внешние клиенты открываются в 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

engine.dispose() при завершении — это и есть graceful shutdown persistence (PYBOOT-12). SIGTERM uvicorn/gunicorn обработает сам: lifespan-блок finally (или часть после yield) выполняется всегда.

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

PYBOOT-8/R-HEX-BOOT-1 — container.py собирает все зависимости. Пример с dependency-injector:

# app/container.py
from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject
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,
        },
    )

Container — источник правды о том, какая реализация подключена к какому порту. SqlAlchemyOrderRepository удовлетворяет Protocol OrderRepository; SberPaymentAdapter — Protocol PaymentPort. Это и есть R-HEX-BOOT-3 на Python: явный wiring вместо component scan.

PYBOOT-9 — недетерминированные источники (clock, id_generator) за Protocol:

    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) — детерминированный результат без патчинга.

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

R-HEX-AIN-2 — роутер маппит DTO в команду и зовёт Dispatcher. Зависимости через Depends, не через глобальный контейнер:

# adapters/in_/http/order_router.py
from fastapi import APIRouter, Depends
from service.core.order.usecase.create_order import CreateOrder
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

Роутер не знает про SqlAlchemyOrderRepository или SberPaymentAdapter — только про Dispatcher. Это R-HEX-AIN-4.

Persistence-wiring

PYBOOT-10/PYBOOT-11 — async SQLAlchemy в lifespan, Alembic для миграций:

# Локальный старт
docker compose up -d postgres
alembic upgrade head
uvicorn service.app.main:app --reload

PYBOOT-X4 — Base.metadata.create_all() в проде запрещён. Миграции — только Alembic, и только в CI/деплое, не в lifespan.

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

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──→ ∅

R-HEX-MOD-X2 — core/ не импортирует adapters/* или app/. R-HEX-AIN-X4 — adapters/in/* не импортирует adapters/out/*. Контракт проверяется import-linter в CI (R-HEX-TEST-1):

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

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

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

АнтипаттернПравилоЧто взамен
Роутер или Handler в app/main.py или app/container.pyR-HEX-BOOT-X1Роутер — в adapters/in/http/, handler — в core/<bc>/usecase/
create_app() или wiring в core/ или adapters/R-HEX-BOOT-X2Только в app/
engine = create_async_engine(...) в глобале модуляPYBOOT-X2В lifespan, передавать через container override
os.getenv("DB_URL") в нескольких местах кодаPYBOOT-X1Один Settings-объект с BaseSettings
from service.app.container import container в core/R-HEX-MOD-X2Стрелка идёт только внутрь: app → adapters → core
Бизнес-логика в app/container.py (if settings.env == "local" → другой handler)R-HEX-BOOT-X1Профильное поведение — через порт с разными реализациями
lifespan или create_app без покрытия import-linterR-HEX-TEST-X1lint-imports как required check в 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
  • Когда переходить — когда Hexagonal оправдан на Python, признаки Уровня 3