← Back to the section

Once core/ and the adapters are ready, they still need to be connected: tell the application which repository to plug into which port, where to get the settings, how to start the server and how to stop it properly. All this "assembly" code lives in one place — app/. Let's look at why it's arranged this way and how it looks in practice.

Why a composition root is needed at all

Imagine that in every code file there's db_url = os.getenv("DB_URL") — and so in ten places. Or that create_async_engine(...) is created on module import rather than at application startup. In such a situation tests are hard to isolate, settings are scattered across the whole codebase, and nowhere is there a full picture of how the service is assembled.

The composition root is the single place in the application where all dependencies are assembled together. It knows about core/, about all adapters, about the settings. But everything else — core/, the adapters, the routers — knows nothing about app/.

In hexagonal architecture on Python that place is the app/ folder.

What lives in app/

The service structure looks like this:

src/<service>/
  app/
    __init__.py
    main.py          # create_app() + launch entry point
    container.py     # DI wiring: Handlers, repositories, Dispatcher
    lifespan.py      # @asynccontextmanager: engine, clients, pools
    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

There are no routers and no domain classes in app/ — only wiring. The router lives in adapters/in/http/, the business logic — in core/.

Settings: a single place for configuration

A common problem is "scattered" settings: os.getenv("DB_URL") in one file, os.getenv("API_KEY") in another. If a variable isn't set, you find out at the worst possible moment — when a request is already being served.

The solution: one Settings class via pydantic-settings. All variables are declared explicitly, typed, and validated right at startup — the application won't start until all the required variables are set.

# 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 and sber_api_key are declared without a default value — these are required fields. If the environment variable isn't set, pydantic fails at startup with a clear message, rather than in the middle of handling a request.

The environment profile (local, integration-test, production) is set only through the APP_ENV environment variable — not by code, not by conditions inside the application.

App factory: a factory instead of a global

Another common mistake is creating the application right at the module level:

# Don't do this
app = FastAPI()
app.include_router(order_router)

The problem: when tests run, this code executes on import, with all its global dependencies. Running two tests with different settings is impossible.

The correct approach is a factory function 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()

The test calls create_app(FakeSettings(...)) and gets a fully isolated application instance. make_router(container) — the router receives its dependencies explicitly through a parameter, not through a global import.

Lifespan: resource management

A database connection, a connection pool, an HTTP client to an external service — these resources need to be opened at startup and closed at shutdown. If you create them on module import, you lose control over the order of initialization and release.

FastAPI solves this via lifespan — a special context manager:

# 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

Everything before yield is startup. Everything after yield is shutdown. On receiving SIGTERM, uvicorn/gunicorn will finish handling the current requests and run the block after yield. engine.dispose() closes the database connection pool.

The DI container: wiring core and adapters

container.py is a dependency map: which adapter is plugged into which port. An example with the dependency-injector library:

# 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 implements the Protocol OrderRepository; SberPaymentAdapter implements the Protocol PaymentPort. core/ doesn't know about the concrete adapter classes — only about the Protocol interfaces.

Non-deterministic sources (clock, id_generator) are also wrapped into providers:

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

In a test, container.clock.override(lambda: fixed_datetime) gives a deterministic result without patching global functions.

How the router gets its dependencies

The router doesn't import the container directly — it receives dependencies through a parameter and passes them into 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

The router knows only about Dispatcher — not about SqlAlchemyOrderRepository or SberPaymentAdapter. The implementation details are hidden behind the container.

Dependency structure

The dependency arrows point strictly inward:

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/ ──depends ONLY on stdlib──→ ∅

core/ doesn't import adapters/ or app/. adapters/in/* doesn't import adapters/out/*. This is checked automatically via import-linter in CI:

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

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

If someone accidentally adds from service.app.container import container in core/, CI fails with a clear message.

Common mistakes

A router or Handler in app/main.py or app/container.py. The router belongs to adapters/in/http/, the handler — to core/<bc>/usecase/. In app/ there's only assembly, not logic.

engine = create_async_engine(...) in the module global. The engine must be created in lifespan and passed through a container override — otherwise it's created on import, before the settings are set, and isn't closed at shutdown.

os.getenv("DB_URL") in several places in the code. Set up a single Settings object with BaseSettings and pass it where it's needed.

from service.app.container import container in core/. The dependency arrow points only inward: app → adapters → core. core/ doesn't know about app/.

Business logic in container.py (if settings.env == "local" — a different handler). Profile-specific behavior is moved into a port with different implementations.

Base.metadata.create_all() in lifespan for production. Migrations are managed via Alembic — in CI and in the deploy, not in the application code.

Local run

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

In production gunicorn -k uvicorn.workers.UvicornWorker is recommended — it handles SIGTERM and hands control to lifespan for a clean shutdown.

In short

  • app/ is the composition root: the single place where all dependencies are assembled. No one imports app/ — only app/ imports everything else.
  • Configuration is a single Settings class based on BaseSettings. Required fields have no default: the application fails at startup if a variable isn't set.
  • create_app(settings) is a factory, not a module-level global. Tests call it with test settings and get an isolated instance.
  • Resources (engine, HTTP clients) go in lifespan. Before yield — open, after yield — close. Graceful shutdown happens automatically.
  • container.py is the dependency map: which adapter implements which port. Routers and handlers don't know about the concrete implementations.
  • The architectural contract (core/ → ∅, adapters/ → core/, app/ → everything) is checked by import-linter in CI.
  • Package structure — the core/adapters/app package layout and the import-linter contract
  • Core layer — what lives in core/ and why FastAPI/SQLAlchemy can't go there
  • Ports — Protocol as an outbound port, port exceptions, domain types in signatures
  • Adapters in — routers, the request-DTO to command mapper, the link through the Dispatcher
  • Adapters out — implementing Protocol ports, domain ↔ DTO mapping in the adapter
  • Architecture tests — import-linter: layers + forbidden + independence contracts in CI