Dependency Injection: Wiring Clean Python Architectures

Lifecycle and Resource Management

When using connection pools33 like httpx.AsyncClient or a database engine , developers must ensure resources are cleanly shut down to avoid leaking connections or close unattended connections.

The Composition Root must act as a context manager to guarantee cleanup during application shutdown or failure. We want to centralize the configuration in a typed settings object, not os.getenv calls scattered through the bootstrap:

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

The bootstrap then hands the settings object a single function whose only job is to open the shared resources, wire the adapters, and close the resources again on exit:

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

Three concrete wins over the os.getenv version: pydantic-settings reads .env files, validates types (database_url is a PostgresDsn, stripe_api_key is a SecretStr so it does not leak into logs), and fails the process at startup if a required value is missing99 pydantic-settings provides BaseSettings, which reads from process environment variables and .env files, supports nested configuration with env_nested_delimiter, and validates types at construction time. SecretStr masks the value in logs and tracebacks. . The context manager guarantees that http_client.aclose() runs even when the surrounding process crashes.

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

Test Doubles Without Monkeypatching

Constructor injection lets you swap any port for a hand-written test double, with no unittest.mock.patch55 like unittest.mock.patch and no module-level state. This can be convienent if you need full control over the double, with no magic involved.

A test for OrderProcessorService only needs to express “given a repository that returns this order, and a gateway that returns success, the service returns true and saves the order”:

import pytest
from shop.domain import Order
from shop.ports import OrderRepositoryPort, PaymentGatewayPort
from shop.services import OrderProcessorService

# Hand-written doubles satisfy the port protocols by structural typing.
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


@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

Global Singletons

Using module-level variables or singleton classes to access dependencies is a common pitfall. When a service imports a global instance directly, it couples itself to that specific database client state, and that state survives across tests in the same process.

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

# service
from .db import db_connection

As mentioned in “Why Module-Level State Fails”, two failure modes follow, and both quite hard and painful to debug. The first is state leakage between tests: a fixture that mutates db_connection leaks into the next test unless the fixture explicitly resets it (a form of “test ordering coupling”). The second is lifecycle confusion with async resources. Constructor injection resolves both by giving each instance its own resources and its own scope.

When to Transition to a DI Framework

For small and medium-sized projects, manual composition in a bootstrap_application() function is preferred because the process uses standard Python, remains easy to debug, and provides clear stack tracebacks. Moving to a DI framework might be worth if:

• A service has more than three or four constructor parameters. Mark Seemann’s “Facade Service” pattern88 Mark Seemann, “Refactoring to Aggregate Services” (2010). Introduces the Facade Service pattern as the response to “too many constructor parameters.” The threshold that matters is per-service parameter count, not the total number of services. treats this as the trigger to refactor: extract a cluster of collaborators into a typed facade, then inject the facade.
• A second entry point appears. Once you have two bootstrap_* functions, the manual composition has stopped paying for itself, and a container that both entry points can share is worth the extra dependency.
• Cross-cutting concerns need to reach many services. A logger, a tracer, a request ID, a clock, feature flags. Injecting these into every service constructor is the textbook “constructor over-injection” smell. The fix is not a container; it is contextvars.ContextVar1010 PEP 567 (Python 3.7+) introduced contextvars for per-task context propagation across await boundaries. structlog, opentelemetry, and modern web frameworks use it for request-scoped metadata. or a middleware.

We’ll explore existing frameworks, what do they offer, and what does the production cost look like in the next post.