Dependency Injection: Wiring Clean Python Architectures

Dependency hardcoupling

When classes instantiate their own dependencies, they assume two distinct responsibilities: business logic execution and resource configuration/instantiation.

Consider this common anti-pattern where a service instantiates its own dependencies directly:

# Anti-pattern: hardcoded dependencies
import os

class OrderProcessorService:
    def __init__(self) -> None:
        # Tightly coupled to a concrete Stripe payment adapter
        self._payment_gateway = StripePaymentAdapter(api_key=os.environ["STRIPE_API_KEY"])

Hardcoded dependencies introduce three critical flaws:

1. Untestable in isolation: the service cannot be unit-tested without hitting Stripe or using complex monkeypatching tools.
2. Configuration pollution: the service must know configuration details11 such as API keys that are completely separate from its core domain rules.
3. Rigid lifecycle management: if the payment adapter needs to share an HTTP connection pool22 like httpx.AsyncClient with other classes, sharing becomes difficult because the adapter is instantiated internally.

Why module-level state fails

The naive fix looks attractive. The easiest approach would be to define the database client and the payment adapter as instances in an adapters module, then import them directly into the services that need them. Construction becomes trivial.

An experienced developer knows this approach does not scale. Three failure modes surface as the project grows, and each one is a debugging trap:

• Test ordering coupling. A test that patches db.fetch_order and runs in isolation passes, then fails in CI when a later test imports the same module and inherits the still-patched attribute. Tests must be order-independent; module-level globals make them order-dependent.
• Async resource lifecycle. A module-level httpx.AsyncClient binds its transport to whichever event loop first uses it. pytest-asyncio creates a fresh loop per test by default, so the client’s transport points at a closed loop on the second test, and every async call raises RuntimeError: Event loop is closed.
• Circular import landmines. A service will eventually raise an ImportError when importing different adapters. Module-level state is a dead end. Dependencies must stay local to each instance and be injected at the boundary.

The Hexagonal Architecture post covered the two opposing directions at the core of the inversion: static dependency (imports point inward from adapters to ports) and runtime control (execution calls flow outward from services to adapters). Dependency injection is the mechanism that resolves the architectural conflict at runtime, by having the Composition Root instantiate the adapters and push them into the services that depend on the corresponding ports.

The composition root

To decouple configuration from execution, there are two structural patterns. Constructor dependency injection declares all dependencies as arguments in the constructor (__init__) and types them using protocols. This is the pattern seen in previous posts when declaring protocols. Then there is the composition root, which establishes a single, centralized location in the application startup flow where the bootstrapper instantiates, configures, and wires all dependencies.

The Composition Root is the single function at the entrypoint of the application where the real adapters are instantiated, configured with concrete values from a settings object, and handed to the services that depend on them. Every other module in the codebase stays free of os.environ reads, httpx.AsyncClient() constructors, and PostgresOrderRepository(...) calls; only the Composition Root does that work.

The function takes a typed settings object (covered below), branches on the environment, and returns a container that holds the wired service alongside any resources whose lifecycle the container must manage. The same composition function can be called from a FastAPI lifespan, a Typer CLI, a Celery worker, or a test fixture, and the resulting OrderProcessorService is identical in every case because the wiring is the same.

Use cases and pitfalls

from pydantic import PostgresDsn, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class AppSettings(BaseSettings):
    # reads from process env + .env
    model_config = SettingsConfigDict(env_file=".env", env_nested_delimiter="__")

    env: str = "development"
    database_url: PostgresDsn
    stripe_api_key: SecretStr
    http_timeout_seconds: float = 5.0

from contextlib import asynccontextmanager

class ApplicationContainer:
    def __init__(self, repository, payment_gateway, http_client=None):
        # ...

@asynccontextmanager
async def application_lifecycle(settings: AppSettings):
    http_client = httpx.AsyncClient(...)  # shared pool
    container = ApplicationContainer(
        repository=PostgresOrderRepository(...),
        payment_gateway=StripePaymentAdapter(http_client=http_client, ...),
        http_client=http_client,
    )
    try:
        yield container
    finally:
        await http_client.aclose()

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = AppSettings()
    async with application_lifecycle(settings=settings) as container:
        app.state.container = container
        yield

@cli.command()
def process_order(order_id: str) -> None:
    async def run() -> None:
        settings = AppSettings(env="production")
        async with application_lifecycle(settings=settings) as container:
            result = await container.order_service.execute(order_id=order_id)
            typer.echo(f"processed={result}")
    asyncio.run(run())

from shop.domain import Order
from shop.ports import OrderRepositoryPort, PaymentGatewayPort

class FakeOrderRepository(OrderRepositoryPort):
    def __init__(self) -> None:
        self.orders: dict[str, Order] = {}
        self.saved_order: Order | None = None

    async def get_by_id(self, order_id: str) -> Order | None:
        return self.orders.get(order_id)

    async def save(self, order: Order) -> None:
        self.saved_order = order


class FakePaymentGateway(PaymentGatewayPort):
    def __init__(self, should_succeed: bool) -> None:
        self.should_succeed = should_succeed

    async def charge(self, order: Order) -> bool:
        return self.should_succeed

import pytest
from shop.domain import Order
from shop.services import OrderProcessorService

@pytest.mark.asyncio
async def test_order_processor_success() -> None:
    repo = FakeOrderRepository()
    repo.orders["order_123"] = Order(id="order_123", customer_email="buyer@test.com", total_amount=150.00)
    gateway = FakePaymentGateway(should_succeed=True)

    service = OrderProcessorService(repository=repo, payment_gateway=gateway)
    result = await service.execute(order_id="order_123")

    assert result is True
    assert repo.saved_order is not None

# db
db_connection = PostgresOrderRepository(connection_string="...")

# service
from .db import db_connection

The DI tool landscape

The dishka alternatives page surveys the current Python DI tooling:

Each tool has its place, so you’ll have to study their unique characteristics to figure out the best fit for your use case. A Celery worker calls for wireup or dishka (both have official Celery integrations) and saves you from rewriting the composition root. A CLI calls for wireup (Typer and Click coverage out of the box). FastAPI-only setups get a working container from Depends.

In general, the most complete framework I’ve evaluated is dishka, which offers some interesting features:

• Scopes (APP, REQUEST, ACTION, STEP, plus SESSION for WebSockets) let each entry point’s lifetime be a first-class concept. FastAPI Depends only has REQUEST; the WebSocket case is hand-rolled.
• Fail-fast graph validation runs at make_container(...) time. A cycle, missing provider, or unsupplied context value raises on the line that imports the providers, not on the first request that exercises the misconfiguration.
• Async finalization is handled by AsyncContainer.__aexit__. A provider declared as async def get_conn(self) -> AsyncIterator[Connection]: ... has its cleanup awaited when the REQUEST scope exits.
• from_context declares ambient data (Request, WebSocket, the authenticated user) once via from_context(Request, scope=Scope.REQUEST). Middleware supplies it at scope entry, and the framework resolves it in any service constructor without parameter plumbing. Depends(get_request) would thread the value through every service that needs it.

What this pattern costs

A Composition Root earns its overhead for an application with more than one entry point. But that’s not always the case:

1. Compose manually for a single entry point with one resource scope. A @asynccontextmanager around the FastAPI lifespan covers it.
2. Adopt a framework when there are multiple entry points with different resource lifecycles. dishka and wireup feed the same provider graph into all entry points; manual composition requires rewriting bootstrap_application per entry point.
3. Instrument the bootstrap in any case. The Composition Root is the single point of failure for misconfiguration (bad database_url, missing stripe_api_key), so put structured startup logging at the top to find the faulting configuration fast.

The next post explores cross-cutting concerns (request IDs, trace IDs, the authenticated user, the current locale) without bloating the services’ constructors, using the contextvars.ContextVar primitive.

References and additional resources

• FastAPI: dependency injection tutorial, dependencies with yield
• Pydantic: Settings
• Mark Seemann’s blog: Service Locator is an Anti-Pattern (2010), Refactoring to Aggregate Services (2010)
• dishka: https://dishka.readthedocs.io
• svcs: https://github.com/hynek/svcs
• wireup integrations: https://maldoinc.github.io/wireup/latest/integrations/