Python Dependency Injection: Decoupled Architecture

Dependency injection (DI) is the practice of passing dependencies into a class or function rather than creating them internally. It makes code testable (swap real dependencies for mocks), configurable (swap implementations without changing callers), and maintainable (dependencies are explicit). Python doesn't require a heavy DI framework — constructor injection and FastAPI's Depends system cover most production needs elegantly.

Why Dependency Injection?

# WITHOUT DI — hard to test, hard to swap implementations
class OrderService:
    def __init__(self):
        self.db = PostgresDatabase("postgresql://localhost/orders")  # hardcoded!
        self.email = SmtpEmailSender("smtp.gmail.com", 587)          # hardcoded!
        self.payment = StripePaymentGateway(api_key="sk_live_...")   # hardcoded!

    def place_order(self, order):
        # Impossible to test without real DB, SMTP server, and Stripe
        pass

# WITH DI — testable, configurable, flexible
class OrderService:
    def __init__(self, db: Database, email: EmailSender, payment: PaymentGateway):
        self.db = db
        self.email = email
        self.payment = payment

    def place_order(self, order):
        # Works with any implementation of Database, EmailSender, PaymentGateway
        pass

# Production: inject real dependencies
service = OrderService(
    db=PostgresDatabase(os.environ["DATABASE_URL"]),
    email=SmtpEmailSender(os.environ["SMTP_HOST"]),
    payment=StripePaymentGateway(os.environ["STRIPE_KEY"]),
)

# Tests: inject fakes
service = OrderService(
    db=InMemoryDatabase(),
    email=FakeEmailSender(),
    payment=FakePaymentGateway(),
)

Constructor Injection

Constructor injection is the cleanest DI approach — all dependencies are explicit in the __init__ signature. Pair it with Python's Protocol (structural subtyping) instead of ABC inheritance for flexible interfaces.

from typing import Protocol, runtime_checkable
from abc import abstractmethod

# Define interfaces as Protocols — no inheritance required
@runtime_checkable
class UserRepository(Protocol):
    async def find_by_id(self, user_id: int) -> dict | None: ...
    async def save(self, user: dict) -> dict: ...
    async def delete(self, user_id: int) -> bool: ...

@runtime_checkable
class EmailService(Protocol):
    async def send(self, to: str, subject: str, body: str) -> None: ...

# Service depends on protocols (interfaces), not concrete classes
class UserService:
    def __init__(self, users: UserRepository, email: EmailService):
        self._users = users
        self._email = email

    async def register(self, username: str, email_addr: str) -> dict:
        user = await self._users.save({"username": username, "email": email_addr})
        await self._email.send(email_addr, "Welcome!", f"Hi {username}, welcome!")
        return user

    async def deactivate(self, user_id: int) -> bool:
        user = await self._users.find_by_id(user_id)
        if not user:
            return False
        await self._email.send(user["email"], "Account Deactivated", "Your account has been deactivated.")
        return await self._users.delete(user_id)

# Concrete implementations
class PostgresUserRepository:
    def __init__(self, pool): self.pool = pool
    async def find_by_id(self, user_id: int): ...
    async def save(self, user: dict): ...
    async def delete(self, user_id: int): ...

class SendgridEmailService:
    def __init__(self, api_key: str): self.api_key = api_key
    async def send(self, to, subject, body): ...

FastAPI Depends System

FastAPI's Depends() is built-in dependency injection that supports sync and async functions, generators (for cleanup), caching per request, and nested dependencies. It's the most pythonic DI approach for FastAPI applications.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
import asyncpg

app = FastAPI()

# Simple dependency
async def get_db_pool() -> asyncpg.Pool:
    return app.state.db_pool  # set in lifespan

async def get_db(pool: asyncpg.Pool = Depends(get_db_pool)) -> asyncpg.Connection:
    async with pool.acquire() as conn:
        yield conn  # cleanup happens automatically after request

# Repository as a dependency
class UserRepo:
    def __init__(self, db: asyncpg.Connection):
        self.db = db

    async def find(self, user_id: int) -> dict | None:
        row = await self.db.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
        return dict(row) if row else None

def get_user_repo(db=Depends(get_db)) -> UserRepo:
    return UserRepo(db)

# Service as a dependency (nesting)
class UserService:
    def __init__(self, repo: UserRepo, email: EmailService):
        self.repo = repo
        self.email = email

def get_email_service() -> EmailService:
    return SendgridEmailService(os.environ["SENDGRID_KEY"])

def get_user_service(
    repo: UserRepo = Depends(get_user_repo),
    email: EmailService = Depends(get_email_service),
) -> UserService:
    return UserService(repo, email)

# Routes use the service
@app.get("/users/{user_id}")
async def get_user(user_id: int, svc: UserService = Depends(get_user_service)):
    user = await svc.repo.find(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.post("/users/{user_id}/deactivate")
async def deactivate(user_id: int, svc: UserService = Depends(get_user_service)):
    ok = await svc.deactivate(user_id)
    return {"success": ok}

DI Containers: dependency-injector

pip install dependency-injector

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

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

    # Infrastructure
    db_pool = providers.Singleton(
        asyncpg.create_pool,
        dsn=config.database.url,
        min_size=config.database.pool_min,
        max_size=config.database.pool_max,
    )

    redis = providers.Singleton(
        aioredis.from_url,
        config.redis.url,
    )

    # Repositories
    user_repo = providers.Factory(
        PostgresUserRepository,
        pool=db_pool,
    )

    # Services
    email_service = providers.Singleton(
        SendgridEmailService,
        api_key=config.email.sendgrid_key,
    )

    user_service = providers.Factory(
        UserService,
        users=user_repo,
        email=email_service,
    )

# Application entry point
container = Container()
container.config.from_yaml("config.yaml")
container.wire(modules=[__name__])

@app.get("/users/{user_id}")
@inject
async def get_user(
    user_id: int,
    svc: UserService = Provide[Container.user_service],
):
    return await svc.repo.find(user_id)

Testing with DI

import pytest
from unittest.mock import AsyncMock, MagicMock

# Fake implementations for tests
class FakeUserRepository:
    def __init__(self):
        self._data = {}
        self._next_id = 1

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

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

    async def delete(self, user_id: int) -> bool:
        return self._data.pop(user_id, None) is not None

class FakeEmailService:
    def __init__(self):
        self.sent: list[dict] = []

    async def send(self, to, subject, body):
        self.sent.append({"to": to, "subject": subject, "body": body})

@pytest.fixture
def user_service():
    repo = FakeUserRepository()
    email = FakeEmailService()
    return UserService(repo, email), repo, email

@pytest.mark.asyncio
async def test_register_sends_welcome_email(user_service):
    svc, repo, email = user_service
    user = await svc.register("alice", "alice@example.com")
    assert user["username"] == "alice"
    assert len(email.sent) == 1
    assert email.sent[0]["to"] == "alice@example.com"
    assert "Welcome" in email.sent[0]["subject"]

# FastAPI override for integration tests
from fastapi.testclient import TestClient

fake_email = FakeEmailService()

def override_get_email_service():
    return fake_email

app.dependency_overrides[get_email_service] = override_get_email_service
client = TestClient(app)

def test_deactivate_endpoint():
    resp = client.post("/users/1/deactivate")
    assert resp.status_code == 200

Common Patterns

# Service Locator (anti-pattern to be aware of)
# Avoid this — hides dependencies, hard to test
class ServiceLocator:
    _services = {}
    @classmethod
    def register(cls, name, instance): cls._services[name] = instance
    @classmethod
    def get(cls, name): return cls._services[name]

# Factory pattern for creating configured objects
class DatabaseFactory:
    @staticmethod
    def create(env: str):
        if env == "test":
            return SQLiteDatabase(":memory:")
        elif env == "prod":
            return PostgresDatabase(os.environ["DATABASE_URL"])
        raise ValueError(f"Unknown env: {env}")

# Scoped dependencies (per-request vs singleton)
# FastAPI: use_cache=True (default) = singleton per request
# use_cache=False = new instance per injection point
@app.get("/data")
async def data(
    db1=Depends(get_db, use_cache=True),   # same instance
    db2=Depends(get_db, use_cache=True),   # same instance
    db3=Depends(get_db, use_cache=False),  # new instance
):
    pass

Frequently Asked Questions

Do I need a DI framework in Python?

Not for most applications. Constructor injection with Python's duck typing and Protocols is clean and effective without any library. Use a DI framework like dependency-injector only when you have many layers of dependencies with complex lifecycles (singleton vs factory vs scoped).

How is DI different from the Service Locator pattern?

With DI, dependencies are pushed into the consumer from the outside (explicit). With Service Locator, the consumer pulls dependencies from a central registry (implicit). DI is preferred because dependencies are visible in function signatures — making the code easier to test and understand without reading the implementation.

How do I handle circular dependencies?

Circular dependencies indicate a design problem — A depends on B which depends on A. Resolve them by extracting a third component C that both A and B depend on, or by introducing an event/message between them. Don't work around circular deps by delaying imports.