Python Dependency Injection: Decoupled Architecture

Dependency injection (DI) is the practice of providing a class or function with the objects it needs rather than letting it create them internally. In Python, DI improves testability by making it trivial to substitute real services with mocks, improves flexibility by letting you swap implementations without changing calling code, and improves readability by making all dependencies explicit in function signatures. FastAPI builds DI into its core with the Depends system; for non-FastAPI code, the dependency-injector library or simple constructor injection achieves the same goals.

Table of Contents

Why Dependency Injection?

Without DI, classes create their own dependencies internally. This makes testing hard (you cannot easily replace a real database with a mock) and couples the class to a specific implementation. DI inverts this: the caller provides the dependencies, and the class only declares what it needs through its constructor or function parameters.

from abc import ABC, abstractmethod

# Define interfaces (protocols) for dependencies
class UserRepository(ABC):
    @abstractmethod
    async def get_by_id(self, user_id: int) -> dict | None: ...

    @abstractmethod
    async def save(self, user: dict) -> dict: ...

class EmailService(ABC):
    @abstractmethod
    async def send_welcome(self, email: str, name: str) -> None: ...

# WITHOUT DI — tightly coupled, untestable
class UserServiceBad:
    def __init__(self):
        import asyncpg
        self._db = asyncpg.connect(...)  # hard-coded!
        self._mailer = SmtpEmailService(host="smtp.gmail.com")  # hard-coded!

# WITH DI — loosely coupled, testable
class UserService:
    def __init__(self, repo: UserRepository, mailer: EmailService):
        self._repo = repo
        self._mailer = mailer

    async def register(self, name: str, email: str) -> dict:
        user = await self._repo.save({"name": name, "email": email})
        await self._mailer.send_welcome(email, name)
        return user

    async def get_profile(self, user_id: int) -> dict:
        user = await self._repo.get_by_id(user_id)
        if not user:
            raise ValueError(f"User {user_id} not found")
        return user

Manual Constructor Injection

For many applications, manual DI is sufficient. You create concrete implementations at application startup and wire them together by passing instances into constructors. This is sometimes called "poor man's DI" but is extremely readable and transparent — the entire object graph is visible in one place.

import asyncpg

# Concrete implementations
class PostgresUserRepository(UserRepository):
    def __init__(self, pool: asyncpg.Pool):
        self._pool = pool

    async def get_by_id(self, user_id: int) -> dict | None:
        async with self._pool.acquire() as conn:
            row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
            return dict(row) if row else None

    async def save(self, user: dict) -> dict:
        async with self._pool.acquire() as conn:
            row = await conn.fetchrow(
                "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING *",
                user["name"], user["email"]
            )
            return dict(row)

class SendgridEmailService(EmailService):
    def __init__(self, api_key: str):
        self._api_key = api_key

    async def send_welcome(self, email: str, name: str) -> None:
        # send via Sendgrid API
        print(f"Sending welcome email to {name} at {email}")

# Wiring at startup (application composition root)
async def create_app():
    pool = await asyncpg.create_pool("postgresql://localhost/mydb")
    repo = PostgresUserRepository(pool)
    mailer = SendgridEmailService(api_key="SG.xxx")
    service = UserService(repo=repo, mailer=mailer)
    return service, pool

# Usage
async def main():
    service, pool = await create_app()
    user = await service.register("Alice", "alice@example.com")
    print(user)

FastAPI Depends System

FastAPI's Depends is a first-class DI system built into the framework. Dependency functions are regular Python functions (sync or async) that FastAPI calls automatically, resolves in the correct order, and caches within a request when use_cache=True (the default). Dependencies can themselves have dependencies, forming a tree that FastAPI resolves automatically.

from fastapi import FastAPI, Depends, HTTPException
from functools import lru_cache
import asyncpg

app = FastAPI()

# Settings dependency — singleton via lru_cache
@lru_cache
def get_settings():
    from pydantic_settings import BaseSettings
    class Settings(BaseSettings):
        db_url: str = "postgresql://localhost/mydb"
        sendgrid_key: str = ""
        model_config = {"env_file": ".env"}
    return Settings()

# Database pool dependency
_pool: asyncpg.Pool | None = None

async def get_pool() -> asyncpg.Pool:
    return _pool  # set during lifespan startup

# Repository dependency — created per-request
async def get_user_repo(pool: asyncpg.Pool = Depends(get_pool)) -> UserRepository:
    return PostgresUserRepository(pool)

# Service dependency — composes lower-level dependencies
async def get_user_service(
    repo: UserRepository = Depends(get_user_repo),
    settings=Depends(get_settings),
) -> UserService:
    mailer = SendgridEmailService(settings.sendgrid_key)
    return UserService(repo=repo, mailer=mailer)

# Route — only declares what it needs
@app.post("/users", status_code=201)
async def create_user(
    name: str,
    email: str,
    service: UserService = Depends(get_user_service),
):
    try:
        return await service.register(name, email)
    except ValueError as e:
        raise HTTPException(400, str(e))

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service),
):
    try:
        return await service.get_profile(user_id)
    except ValueError:
        raise HTTPException(404, "User not found")

dependency-injector Library

The dependency-injector library brings a full IoC container to Python. It supports singleton, factory, and resource providers, wires dependencies via decorators, and integrates with FastAPI, Flask, Django, and CLI applications. It's the right choice for large applications with many services and complex dependency graphs.

from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject

class Container(containers.DeclarativeContainer):
    # Configuration
    config = providers.Configuration()

    # Infrastructure
    db_pool = providers.Resource(
        asyncpg.create_pool,
        dsn=config.db_url,
        min_size=5,
        max_size=20,
    )

    # Repositories — factory creates a new instance each time
    user_repo = providers.Factory(
        PostgresUserRepository,
        pool=db_pool,
    )

    # External services — singleton (one instance shared everywhere)
    email_service = providers.Singleton(
        SendgridEmailService,
        api_key=config.sendgrid_key,
    )

    # Application services
    user_service = providers.Factory(
        UserService,
        repo=user_repo,
        mailer=email_service,
    )

# Wiring: inject dependencies into functions automatically
container = Container()
container.config.from_dict({
    "db_url": "postgresql://localhost/mydb",
    "sendgrid_key": "SG.xxx",
})
container.wire(modules=[__name__])

@inject
async def handle_registration(
    name: str,
    email: str,
    service: UserService = Provide[Container.user_service],
):
    return await service.register(name, email)

# FastAPI integration
@app.post("/users")
@inject
async def create_user_endpoint(
    name: str,
    email: str,
    service: UserService = Depends(Provide[Container.user_service]),
):
    return await service.register(name, email)

Testing with Dependency Injection

DI's primary payoff is in testing. Because dependencies are injected rather than hard-coded, you replace them with mocks in tests without any monkey-patching. FastAPI makes this especially clean with app.dependency_overrides, which swaps any dependency function with a test replacement for the duration of a test.

import pytest
from unittest.mock import AsyncMock, MagicMock
from fastapi.testclient import TestClient
from httpx import AsyncClient, ASGITransport

# Mock implementations
class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self._store: dict[int, dict] = {}
        self._next_id = 1

    async def get_by_id(self, user_id: int) -> dict | None:
        return self._store.get(user_id)

    async def save(self, user: dict) -> dict:
        user = {**user, "id": self._next_id}
        self._store[self._next_id] = user
        self._next_id += 1
        return user

class FakeEmailService(EmailService):
    def __init__(self):
        self.sent: list[tuple[str, str]] = []

    async def send_welcome(self, email: str, name: str) -> None:
        self.sent.append((email, name))

# FastAPI dependency override
@pytest.fixture
def test_client():
    fake_repo = InMemoryUserRepository()
    fake_mail = FakeEmailService()
    fake_service = UserService(repo=fake_repo, mailer=fake_mail)

    app.dependency_overrides[get_user_service] = lambda: fake_service
    yield TestClient(app), fake_mail
    app.dependency_overrides.clear()

def test_create_user(test_client):
    client, mailer = test_client
    response = client.post("/users?name=Alice&email=alice@example.com")
    assert response.status_code == 201
    assert response.json()["name"] == "Alice"
    assert len(mailer.sent) == 1
    assert mailer.sent[0] == ("alice@example.com", "Alice")

def test_get_nonexistent_user(test_client):
    client, _ = test_client
    response = client.get("/users/999")
    assert response.status_code == 404

Common DI Patterns

Beyond basic constructor injection, several DI patterns appear repeatedly in production Python codebases: the repository pattern for data access abstraction, the unit of work pattern for transaction management, and the service locator (used sparingly) for cases where the dependency is not known until runtime.

from contextlib import asynccontextmanager
from typing import AsyncIterator

# Unit of Work pattern — groups repository operations in a transaction
class UnitOfWork:
    def __init__(self, pool: asyncpg.Pool):
        self._pool = pool
        self.users: PostgresUserRepository | None = None

    async def __aenter__(self):
        self._conn = await self._pool.acquire()
        self._tx = self._conn.transaction()
        await self._tx.start()
        self.users = PostgresUserRepository(self._conn)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        if exc_type:
            await self._tx.rollback()
        else:
            await self._tx.commit()
        await self._pool.release(self._conn)

# Inject via FastAPI
@asynccontextmanager
async def get_uow(pool=Depends(get_pool)) -> AsyncIterator[UnitOfWork]:
    async with UnitOfWork(pool) as uow:
        yield uow

@app.post("/transfer")
async def transfer(
    from_id: int, to_id: int, amount: float,
    uow: UnitOfWork = Depends(get_uow),
):
    sender = await uow.users.get_by_id(from_id)
    recipient = await uow.users.get_by_id(to_id)
    if not sender or not recipient:
        raise HTTPException(404)
    # Both updates happen in the same transaction
    await uow.users.debit(from_id, amount)
    await uow.users.credit(to_id, amount)
    return {"status": "transferred"}
Python Articles