Clean Code Python: Full-Stack DI with dependency-injector, FastAPI, and SQLAlchemy

Part 5 showed how FastAPI’s Depends() wires repositories and services per-request. That works. But it scatters your object graph across dozens of factory functions in deps.py, and it only runs inside a request context. Need the same service in a CLI script, a background worker, or a migration hook? You are back to manual construction.

The dependency-injector library solves this. It gives you a declarative container — one file that describes every dependency, how it is created, and how long it lives. FastAPI routes pull from that container. So do tests, CLI tools, and workers. The wiring is visible, overridable, and never duplicated.

This post walks through a full ShelfWise feature — from database session to API response — using dependency-injector with async SQLAlchemy 2.0 and FastAPI.

Why dependency-injector on Top of Depends()

FastAPI’s Depends() is not a general-purpose DI container. It is a request-scoped resolver tied to the ASGI lifecycle. That is perfect for route handlers, but it leaves gaps:

The key insight: dependency-injector does not replace Depends(). It feeds into it. Your container creates the objects; FastAPI’s Depends() delivers them to route handlers.

Step 1: Install and Project Structure

pip install dependency-injector[yaml]
# or in pyproject.toml:
# dependencies = ["dependency-injector[yaml]>=4.44,<5"]

The [yaml] extra adds PyYAML support for loading configuration files. If you only use environment variables, skip it.

Here is the project structure we are building toward:

src/
├── containers.py          # ← The single container (new)
├── config.yml             # ← External configuration (new)
├── main.py
├── api/
│   ├── deps.py            # ← Simplified: pulls from container
│   └── v1/
│       └── books.py
├── db/
│   └── session.py
├── models/
│   └── book.py
├── repositories/
│   ├── protocols.py
│   └── book_repository.py
├── services/
│   └── book_service.py
└── core/
    ├── config.py
    └── exceptions.py

The only new files are containers.py and config.yml. Everything else stays where it was.

Step 2: Configuration Provider

Start with configuration. In Part 10, we used pydantic-settings for typed config. dependency-injector adds a layer on top: a Configuration provider that loads from multiple sources and is injectable like any other dependency.

database:
  url: "postgresql+asyncpg://user:pass@localhost:5432/shelfwise"
  pool_size: 20
  max_overflow: 10
  echo: false

app:
  debug: false
  secret_key: "change-me-in-production"

# src/containers.py — Step 2: Configuration only
from dependency_injector import containers, providers


class Container(containers.DeclarativeContainer):
    """ShelfWise dependency container.

    Single source of truth for the entire object graph.
    """

    wiring_config = containers.WiringConfiguration(
        modules=[
            "src.api.v1.books",
            "src.api.deps",
        ],
    )

    config = providers.Configuration(yaml_files=["src/config.yml"])

wiring_config tells the container which modules to inject into. When the container initializes, it scans those modules for @inject decorators and resolves Provide[...] markers automatically.

# Override nested YAML keys with environment variables:
container = Container()
container.config.from_yaml("src/config.yml")
container.config.database.url.from_env("DATABASE_URL")  # explicit override
# Or use the convention: DATABASE__URL env var → config.database.url

Step 3: Database Resource Provider

The database engine and session factory are singletons — created once at startup, shared across all requests. The async session itself is per-request — created fresh, yielded, and cleaned up.

from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)


def create_engine(
    url: str,
    pool_size: int = 20,
    max_overflow: int = 10,
    echo: bool = False,
) -> AsyncEngine:
    return create_async_engine(
        url,
        pool_size=pool_size,
        max_overflow=max_overflow,
        echo=echo,
    )


def create_session_factory(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
    return async_sessionmaker(
        engine,
        class_=AsyncSession,
        expire_on_commit=False,
    )

Now wire these into the container:

# src/containers.py — Step 3: Add database providers
from dependency_injector import containers, providers

from src.db.session import create_engine, create_session_factory


class Container(containers.DeclarativeContainer):

    wiring_config = containers.WiringConfiguration(
        modules=[
            "src.api.v1.books",
            "src.api.deps",
        ],
    )

    config = providers.Configuration(yaml_files=["src/config.yml"])

    # Singleton: one engine for the lifetime of the process
    db_engine = providers.Singleton(
        create_engine,
        url=config.database.url,
        pool_size=config.database.pool_size.as_int(),
        max_overflow=config.database.max_overflow.as_int(),
        echo=config.database.echo.as_(bool),
    )

    # Singleton: one session factory, bound to the engine
    db_session_factory = providers.Singleton(
        create_session_factory,
        engine=db_engine,
    )

    # Factory: new session per call (per request)
    db_session = providers.Factory(
        db_session_factory.provided(),
    )

Three providers, three lifetimes:

Step 4: Repository and Service Providers

Repositories receive a session. Services receive repositories. The container makes this explicit:

# src/containers.py — Step 4: Full container
from dependency_injector import containers, providers

from src.db.session import create_engine, create_session_factory
from src.repositories.book_repository import BookRepository
from src.services.book_service import BookService


class Container(containers.DeclarativeContainer):

    wiring_config = containers.WiringConfiguration(
        modules=[
            "src.api.v1.books",
            "src.api.deps",
        ],
    )

    config = providers.Configuration(yaml_files=["src/config.yml"])

    # --- Database layer ---
    db_engine = providers.Singleton(
        create_engine,
        url=config.database.url,
        pool_size=config.database.pool_size.as_int(),
        max_overflow=config.database.max_overflow.as_int(),
        echo=config.database.echo.as_(bool),
    )

    db_session_factory = providers.Singleton(
        create_session_factory,
        engine=db_engine,
    )

    db_session = providers.Factory(
        db_session_factory.provided(),
    )

    # --- Repository layer ---
    book_repository = providers.Factory(
        BookRepository,
        session=db_session,
    )

    # --- Service layer ---
    book_service = providers.Factory(
        BookService,
        repo=book_repository,
    )

Read this top to bottom and you see the entire object graph. book_service needs a BookRepository. BookRepository needs an AsyncSession. The session comes from a factory bound to a singleton engine. No guessing, no tracing through five files of Depends() chains.

Step 5: Visualizing the Container Graph

The container describes a dependency graph identical in structure to the Depends() chain from Part 5, but declared in one place:

Every Factory provider creates a new instance per call. Every Singleton provider reuses the same instance. The graph resolves lazily — nothing is created until first accessed.

Step 6: Wire the Container into FastAPI

The container initializes at application startup. FastAPI’s lifespan hook is the right place:

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

from fastapi import FastAPI

from src.containers import Container


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Container is already wired via WiringConfiguration
    yield
    # Shutdown: dispose engine to close all pooled connections
    engine = app.state.container.db_engine()
    await engine.dispose()


def create_app() -> FastAPI:
    container = Container()
    container.config.from_yaml("src/config.yml")
    # Environment variables override YAML (production secrets)
    container.config.database.url.from_env("DATABASE_URL", required=False)

    app = FastAPI(
        title="ShelfWise",
        lifespan=lifespan,
    )
    app.state.container = container

    # Wire the container — scans modules listed in WiringConfiguration
    container.wire()

    from src.api.v1.books import router as books_router
    app.include_router(books_router, prefix="/api/v1")

    return app


app = create_app()

container.wire() is the key call. It patches the modules listed in WiringConfiguration, replacing Provide[...] markers with actual provider resolution. After this call, any function decorated with @inject in those modules will receive container-managed dependencies.

Step 7: Inject into Route Handlers

Two approaches work. Choose based on whether you want routes to know about the container.

Approach A: @inject + Provide (Recommended)

Routes declare dependencies using Provide[...] markers. The container resolves them automatically via wiring:

from dependency_injector.wiring import inject, Provide
from fastapi import APIRouter, Depends, HTTPException

from src.containers import Container
from src.schemas.book import BookCreate, BookResponse
from src.services.book_service import BookService
from src.core.exceptions import BookNotFoundError, InvalidISBNError

router = APIRouter(prefix="/books", tags=["books"])


@router.get("/{book_id}", response_model=BookResponse)
@inject
async def get_book(
    book_id: int,
    service: BookService = Depends(Provide[Container.book_service]),
) -> BookResponse:
    try:
        return await service.get_book(book_id)
    except BookNotFoundError:
        raise HTTPException(status_code=404, detail=f"Book {book_id} not found")


@router.post("/", response_model=BookResponse, status_code=201)
@inject
async def create_book(
    data: BookCreate,
    service: BookService = Depends(Provide[Container.book_service]),
) -> BookResponse:
    try:
        return await service.add_book(data)
    except InvalidISBNError as e:
        raise HTTPException(status_code=422, detail=str(e))


@router.get("/", response_model=list[BookResponse])
@inject
async def list_books(
    q: str | None = None,
    service: BookService = Depends(Provide[Container.book_service]),
) -> list[BookResponse]:
    return await service.list_books(query=q)

Depends(Provide[Container.book_service]) bridges the two systems: Provide[...] resolves from the container, Depends() integrates with FastAPI’s request lifecycle. The @inject decorator activates the wiring.

Approach B: Thin deps.py Bridge

If you prefer routes to stay unaware of the container, wrap the provider in a dependency function:

from dependency_injector.wiring import inject, Provide
from fastapi import Depends
from typing import Annotated

from src.containers import Container
from src.services.book_service import BookService


@inject
async def get_book_service(
    service: BookService = Depends(Provide[Container.book_service]),
) -> BookService:
    return service


BookServiceDep = Annotated[BookService, Depends(get_book_service)]

# src/api/v1/books.py — Approach B (routes don't import Container)
from fastapi import APIRouter, HTTPException
from src.api.deps import BookServiceDep
from src.schemas.book import BookCreate, BookResponse
from src.core.exceptions import BookNotFoundError

router = APIRouter(prefix="/books", tags=["books"])


@router.get("/{book_id}", response_model=BookResponse)
async def get_book(book_id: int, service: BookServiceDep) -> BookResponse:
    try:
        return await service.get_book(book_id)
    except BookNotFoundError:
        raise HTTPException(status_code=404, detail=f"Book {book_id} not found")

Approach B keeps the BookServiceDep pattern from Part 5. Routes are identical — only deps.py changes to pull from the container instead of a hand-written factory chain.

Step 8: Session Lifecycle with Yield

The Factory provider creates a new session per call, but who closes it? In vanilla Depends(), a yield dependency handled cleanup. With dependency-injector, use a Resource provider for dependencies that need explicit teardown:

# src/containers.py — Resource provider for session lifecycle
from dependency_injector import containers, providers
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession

from src.db.session import create_engine, create_session_factory
from src.repositories.book_repository import BookRepository
from src.services.book_service import BookService


async def async_session_resource(
    session_factory: async_sessionmaker,
) -> AsyncIterator[AsyncSession]:
    """Yield a session and guarantee cleanup — mirrors the Depends(get_db) pattern."""
    async with session_factory() as session:
        yield session


class Container(containers.DeclarativeContainer):

    wiring_config = containers.WiringConfiguration(
        modules=[
            "src.api.v1.books",
            "src.api.deps",
        ],
    )

    config = providers.Configuration(yaml_files=["src/config.yml"])

    db_engine = providers.Singleton(
        create_engine,
        url=config.database.url,
        pool_size=config.database.pool_size.as_int(),
        max_overflow=config.database.max_overflow.as_int(),
        echo=config.database.echo.as_(bool),
    )

    db_session_factory = providers.Singleton(
        create_session_factory,
        engine=db_engine,
    )

    # Resource: yields a session, cleans up after request
    db_session = providers.Resource(
        async_session_resource,
        session_factory=db_session_factory,
    )

    book_repository = providers.Factory(
        BookRepository,
        session=db_session,
    )

    book_service = providers.Factory(
        BookService,
        repo=book_repository,
    )

The Resource provider calls async_session_resource, yields the session, and runs the cleanup (session.close() via the async with block) after the provider scope ends. When used with FastAPI’s Depends(), the cleanup aligns with the request lifecycle — same behavior as the yield pattern from Part 5, but declared in the container.

Step 9: Testing with Container Overrides

This is where dependency-injector pays for itself. Override any provider — at any depth — without touching application code:

import pytest
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from src.containers import Container
from src.main import create_app


@pytest.fixture
async def test_engine():
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        echo=True,
    )
    # Create tables
    async with engine.begin() as conn:
        from src.models.base import Base
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()


@pytest.fixture
async def test_session_factory(test_engine):
    return async_sessionmaker(
        test_engine,
        class_=AsyncSession,
        expire_on_commit=False,
    )


@pytest.fixture
def container(test_session_factory) -> Container:
    container = Container()
    # Override the session factory — everything downstream uses the test DB
    container.db_session_factory.override(test_session_factory)
    yield container
    container.unwire()
    container.reset_override()


@pytest.fixture
def app(container):
    app = create_app()
    app.state.container = container
    container.wire()
    return app


@pytest.fixture
async def client(app):
    from httpx import ASGITransport, AsyncClient
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac

The critical line is container.db_session_factory.override(test_session_factory). This single override replaces the production Postgres engine with an in-memory SQLite — and every provider downstream (session, repository, service) automatically uses the test database. No monkeypatching, no dependency_overrides dict.

import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_create_book(client: AsyncClient) -> None:
    payload = {
        "title": "Clean Code Python",
        "isbn": "9780134685991",
        "price": "29.99",
    }
    response = await client.post("/api/v1/books/", json=payload)
    assert response.status_code == 201
    assert response.json()["title"] == "Clean Code Python"


@pytest.mark.asyncio
async def test_get_book_not_found(client: AsyncClient) -> None:
    response = await client.get("/api/v1/books/999")
    assert response.status_code == 404


@pytest.mark.asyncio
async def test_list_books_empty(client: AsyncClient) -> None:
    response = await client.get("/api/v1/books/")
    assert response.status_code == 200
    assert response.json() == []

Tests read exactly like Part 5’s tests. The difference is invisible to the test code — it is in the fixture wiring, which is cleaner and more powerful.

Unit Testing Services Without HTTP

Because the container manages object creation outside FastAPI, you can test services directly without spinning up an ASGI app:

import pytest
from src.services.book_service import BookService
from tests.fakes import FakeBookRepository
from src.core.exceptions import BookNotFoundError


@pytest.fixture
def service() -> BookService:
    return BookService(repo=FakeBookRepository())


@pytest.mark.asyncio
async def test_get_book_raises_not_found(service: BookService) -> None:
    with pytest.raises(BookNotFoundError):
        await service.get_book(book_id=9999)

This test does not use the container at all. It constructs the service with a fake repository manually. The container does not prevent simple construction — it just makes it unnecessary in production code and integration tests.

Step 10: Using the Container Outside FastAPI

The killer feature. A CLI script that reuses the exact same services:

"""CLI tool to bulk-import books from a CSV file."""
import asyncio
import csv
import sys

from dependency_injector.wiring import inject, Provide

from src.containers import Container
from src.services.book_service import BookService
from src.schemas.book import BookCreate


@inject
async def import_books(
    csv_path: str,
    service: BookService = Provide[Container.book_service],
) -> None:
    with open(csv_path) as f:
        reader = csv.DictReader(f)
        for row in reader:
            book = BookCreate(
                title=row["title"],
                isbn=row["isbn"],
                price=row["price"],
            )
            await service.add_book(book)
            print(f"Imported: {book.title}")


async def main() -> None:
    container = Container()
    container.config.from_yaml("src/config.yml")
    container.config.database.url.from_env("DATABASE_URL", required=False)
    container.wire(modules=[__name__])

    await import_books(sys.argv[1])

    # Cleanup
    engine = container.db_engine()
    await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

Same BookService, same BookRepository, same session management. No FastAPI, no ASGI, no Depends(). The container handles wiring. This is impossible with vanilla Depends() — you would need to manually construct the entire dependency chain or duplicate the factory functions.

Step 11: Adding More Services

As the application grows, the container grows with it — but stays readable. Here is the container with an OrderService that depends on both BookRepository and OrderRepository:

# src/containers.py — Extended with OrderService
class Container(containers.DeclarativeContainer):

    wiring_config = containers.WiringConfiguration(
        modules=[
            "src.api.v1.books",
            "src.api.v1.orders",
            "src.api.deps",
        ],
    )

    config = providers.Configuration(yaml_files=["src/config.yml"])

    # --- Database ---
    db_engine = providers.Singleton(
        create_engine,
        url=config.database.url,
        pool_size=config.database.pool_size.as_int(),
        max_overflow=config.database.max_overflow.as_int(),
        echo=config.database.echo.as_(bool),
    )

    db_session_factory = providers.Singleton(
        create_session_factory,
        engine=db_engine,
    )

    db_session = providers.Resource(
        async_session_resource,
        session_factory=db_session_factory,
    )

    # --- Repositories ---
    book_repository = providers.Factory(
        BookRepository,
        session=db_session,
    )

    order_repository = providers.Factory(
        OrderRepository,
        session=db_session,
    )

    # --- Services ---
    book_service = providers.Factory(
        BookService,
        repo=book_repository,
    )

    order_service = providers.Factory(
        OrderService,
        book_repo=book_repository,
        order_repo=order_repository,
    )

Both book_repository and order_repository depend on db_session. The container ensures they receive the same session instance within a single resolution — just like FastAPI’s Depends() deduplication, but explicit and visible.

When Not to Use dependency-injector

dependency-injector adds a layer. That layer must earn its place:

For ShelfWise — a multi-tenant SaaS with CLI tools, background workers, and a growing service layer — the container pays for itself by Part 14. For a weekend project with three endpoints, it is overhead.

The Full Request Flow

To tie everything together, here is what happens when GET /api/v1/books/42 hits the server:

Every box in this diagram maps to a named provider in the container. The entire resolution chain is traceable from containers.py without reading any other file.

Key Takeaways

• One file, one truth — containers.py declares the full object graph. No scattered factory functions, no hidden globals.
• Three provider types matter most — Singleton for process-lifetime objects (engine, config), Factory for per-call objects (services, repositories), Resource for objects that need cleanup (database sessions).
• Container feeds Depends() — Depends(Provide[Container.x]) bridges the container into FastAPI’s request lifecycle. Both systems cooperate; neither replaces the other.
• Override at any depth — container.db_session_factory.override(...) swaps the database for every downstream provider in one line. Tests are cleaner than dependency_overrides.
• Works outside FastAPI — CLI scripts, background workers, and migration hooks use the same container and the same services. No duplication.
• Adopt incrementally — Start with configuration and database providers. Move services into the container as the graph grows. Vanilla Depends() and container providers can coexist in the same application.

Next: the series continues to build on ShelfWise. Every pattern from Parts 1–23 works unchanged with a container — the container just makes the wiring explicit and reusable.