Dependency Injection - Decoupling Components

Both functions below send a welcome email after user registration. One is trivially testable; the other requires monkey-patching or spinning up an SMTP server. Which is which?

# Version A
import smtplib

def register_user_a(email: str, password: str) -> dict:
    user = create_user_in_db(email, password)
    server = smtplib.SMTP("smtp.company.com", 587)
    server.starttls()
    server.login("[email protected]", "secret123")
    server.sendmail("[email protected]", email, "Welcome!")
    server.quit()
    return {"id": user.id}

# Version B
from typing import Protocol

class EmailSender(Protocol):
    def send(self, to: str, subject: str, body: str) -> None: ...

def register_user_b(
    email: str, password: str, email_sender: EmailSender
) -> dict:
    user = create_user_in_db(email, password)
    email_sender.send(email, "Welcome", "Welcome to our platform!")
    return {"id": user.id}

Version B accepts its dependency as a parameter. In tests, you pass a fake. In production, you pass the real SMTP sender. That is dependency injection in its simplest form - and it changes how you design entire systems.

What You Will Learn

• Why dependency injection matters for testability, flexibility, and maintainability
• Manual DI through constructor injection and function parameters
• Building a simple DI container from scratch
• Using the dependency-injector library for production applications
• FastAPI's Depends system as a framework-specific DI mechanism
• The DI pyramid: pure functions, injected services, framework glue
• When DI adds value and when it introduces unnecessary complexity

Prerequisites

• Understanding of Clean Architecture and Hexagonal Architecture (previous two lessons)
• Familiarity with typing.Protocol for defining interfaces
• Experience with FastAPI route handlers and the Depends function
• Comfort with Python classes, closures, and decorators

Part 1 - The Problem DI Solves

Without dependency injection, components create their own dependencies. This creates tight coupling.

# Tight coupling: OrderService creates its own dependencies
class OrderService:
    def __init__(self):
        self._repo = PostgresOrderRepository("postgresql://localhost/db")
        self._payment = StripePaymentProcessor("sk_live_xxx")
        self._email = SendGridEmailSender("SG.xxx")

    def place_order(self, user_id: int, items: list) -> Order:
        order = Order(user_id=user_id, items=items)
        self._payment.charge(order.total)
        self._repo.save(order)
        self._email.send_confirmation(order)
        return order

Testing this requires a running PostgreSQL, valid Stripe credentials, and a SendGrid account. Swapping PostgreSQL for DynamoDB means rewriting OrderService.

With DI, OrderService receives its dependencies from the outside:

# Loose coupling: dependencies injected via constructor
class OrderService:
    def __init__(
        self,
        repo: OrderRepository,       # Protocol
        payment: PaymentProcessor,    # Protocol
        email: EmailSender,           # Protocol
    ):
        self._repo = repo
        self._payment = payment
        self._email = email

    def place_order(self, user_id: int, items: list) -> Order:
        order = Order(user_id=user_id, items=items)
        self._payment.charge(order.total)
        self._repo.save(order)
        self._email.send_confirmation(order)
        return order

The service does not know or care what concrete classes fulfill those protocols.

Part 2 - Manual DI (Constructor Injection)

Constructor injection is the simplest and most common form of DI. You pass dependencies through __init__.

The Pattern

from typing import Protocol, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4


# --- Ports (interfaces) ---

class UserRepository(Protocol):
    def get(self, user_id: UUID) -> Optional["User"]: ...
    def save(self, user: "User") -> None: ...


class PasswordHasher(Protocol):
    def hash(self, plain: str) -> str: ...
    def verify(self, plain: str, hashed: str) -> bool: ...


class AuditLogger(Protocol):
    def log(self, event: str, user_id: UUID, details: str) -> None: ...


# --- Domain entity ---

@dataclass
class User:
    id: UUID
    email: str
    hashed_password: str
    is_active: bool = True


# --- Service with constructor injection ---

class AuthService:
    """All dependencies injected through the constructor."""

    def __init__(
        self,
        user_repo: UserRepository,
        hasher: PasswordHasher,
        audit: AuditLogger,
    ) -> None:
        self._user_repo = user_repo
        self._hasher = hasher
        self._audit = audit

    def authenticate(self, email: str, password: str) -> Optional[User]:
        # In production this would look up by email; simplified for clarity
        user = self._user_repo.get_by_email(email)
        if user is None:
            self._audit.log("auth_failed", uuid4(), f"Unknown email: {email}")
            return None

        if not self._hasher.verify(password, user.hashed_password):
            self._audit.log("auth_failed", user.id, "Wrong password")
            return None

        self._audit.log("auth_success", user.id, "Logged in")
        return user

    def change_password(self, user_id: UUID, old: str, new: str) -> bool:
        user = self._user_repo.get(user_id)
        if user is None:
            return False
        if not self._hasher.verify(old, user.hashed_password):
            self._audit.log("password_change_failed", user_id, "Wrong old password")
            return False

        user.hashed_password = self._hasher.hash(new)
        self._user_repo.save(user)
        self._audit.log("password_changed", user_id, "Password updated")
        return True

Wiring Manually (Composition Root)

# main.py - the composition root
from adapters.postgres_user_repo import PostgresUserRepository
from adapters.bcrypt_hasher import BcryptPasswordHasher
from adapters.file_audit_logger import FileAuditLogger


def create_auth_service() -> AuthService:
    return AuthService(
        user_repo=PostgresUserRepository(session_factory),
        hasher=BcryptPasswordHasher(rounds=12),
        audit=FileAuditLogger("/var/log/app/audit.log"),
    )

Testing With Fakes

# tests/test_auth_service.py
from uuid import uuid4

class FakeUserRepository:
    def __init__(self):
        self._users: dict[UUID, User] = {}

    def get(self, user_id: UUID) -> Optional[User]:
        return self._users.get(user_id)

    def get_by_email(self, email: str) -> Optional[User]:
        for u in self._users.values():
            if u.email == email:
                return u
        return None

    def save(self, user: User) -> None:
        self._users[user.id] = user

    def add(self, user: User) -> None:
        """Test helper to pre-populate."""
        self._users[user.id] = user


class FakeHasher:
    def hash(self, plain: str) -> str:
        return f"hashed:{plain}"

    def verify(self, plain: str, hashed: str) -> bool:
        return hashed == f"hashed:{plain}"


class FakeAuditLogger:
    def __init__(self):
        self.entries: list[tuple] = []

    def log(self, event: str, user_id: UUID, details: str) -> None:
        self.entries.append((event, user_id, details))


def test_authenticate_success():
    repo = FakeUserRepository()
    hasher = FakeHasher()
    audit = FakeAuditLogger()

    user = User(id=uuid4(), email="[email protected]", hashed_password="hashed:secret")
    repo.add(user)

    service = AuthService(user_repo=repo, hasher=hasher, audit=audit)
    result = service.authenticate("[email protected]", "secret")

    assert result is not None
    assert result.email == "[email protected]"
    assert audit.entries[-1][0] == "auth_success"


def test_authenticate_wrong_password():
    repo = FakeUserRepository()
    hasher = FakeHasher()
    audit = FakeAuditLogger()

    user = User(id=uuid4(), email="[email protected]", hashed_password="hashed:secret")
    repo.add(user)

    service = AuthService(user_repo=repo, hasher=hasher, audit=audit)
    result = service.authenticate("[email protected]", "wrong")

    assert result is None
    assert audit.entries[-1][0] == "auth_failed"


def test_change_password():
    repo = FakeUserRepository()
    hasher = FakeHasher()
    audit = FakeAuditLogger()

    user_id = uuid4()
    user = User(id=user_id, email="[email protected]", hashed_password="hashed:old_pass")
    repo.add(user)

    service = AuthService(user_repo=repo, hasher=hasher, audit=audit)
    result = service.change_password(user_id, "old_pass", "new_pass")

    assert result is True
    assert repo.get(user_id).hashed_password == "hashed:new_pass"

:::tip Constructor Injection Scales Further Than You Think Many codebases reach for DI containers too early. Manual constructor injection works well for projects with up to 20-30 services. The composition root gets longer, but it is explicit and debuggable. :::

Part 3 - Building a Minimal DI Container

To understand what DI containers do, let us build one from scratch.

# container.py - a minimal DI container in ~50 lines
from typing import Any, Callable, TypeVar

T = TypeVar("T")


class Container:
    """
    A simple DI container that stores factory functions
    and manages singleton vs transient lifetimes.
    """

    def __init__(self) -> None:
        self._factories: dict[type, Callable] = {}
        self._singletons: dict[type, Any] = {}
        self._singleton_types: set[type] = set()

    def register(
        self,
        interface: type[T],
        factory: Callable[..., T],
        singleton: bool = False,
    ) -> None:
        """Register a factory function for an interface type."""
        self._factories[interface] = factory
        if singleton:
            self._singleton_types.add(interface)

    def resolve(self, interface: type[T]) -> T:
        """Resolve an interface to a concrete instance."""
        if interface in self._singletons:
            return self._singletons[interface]

        if interface not in self._factories:
            raise KeyError(f"No factory registered for {interface}")

        instance = self._factories[interface](self)

        if interface in self._singleton_types:
            self._singletons[interface] = instance

        return instance

    def reset(self) -> None:
        """Clear all singletons (useful for testing)."""
        self._singletons.clear()

Using the Container

# Usage
container = Container()

# Register factories - each receives the container to resolve sub-dependencies
container.register(
    UserRepository,
    lambda c: PostgresUserRepository(session_factory),
    singleton=False,  # new instance per resolve
)

container.register(
    PasswordHasher,
    lambda c: BcryptPasswordHasher(rounds=12),
    singleton=True,  # shared instance
)

container.register(
    AuditLogger,
    lambda c: FileAuditLogger("/var/log/app/audit.log"),
    singleton=True,
)

container.register(
    AuthService,
    lambda c: AuthService(
        user_repo=c.resolve(UserRepository),
        hasher=c.resolve(PasswordHasher),
        audit=c.resolve(AuditLogger),
    ),
    singleton=False,
)

# Resolve
auth = container.resolve(AuthService)
# auth receives PostgresUserRepository, BcryptPasswordHasher, FileAuditLogger

Part 4 - The dependency-injector Library

For production applications, the dependency-injector library provides a robust, well-tested container with features like configuration, singletons, factories, and wiring.

pip install dependency-injector

Defining a Container

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

from adapters.postgres_user_repo import PostgresUserRepository
from adapters.bcrypt_hasher import BcryptPasswordHasher
from adapters.file_audit_logger import FileAuditLogger
from services.auth_service import AuthService
from services.order_service import OrderService


class AppContainer(containers.DeclarativeContainer):
    """Application-level DI container."""

    # Configuration
    config = providers.Configuration()

    # Infrastructure (singletons)
    db_session_factory = providers.Singleton(
        create_session_factory,
        db_url=config.database.url,
    )

    # Repositories (factory - new instance per request)
    user_repo = providers.Factory(
        PostgresUserRepository,
        session_factory=db_session_factory,
    )

    # Utilities (singleton)
    password_hasher = providers.Singleton(
        BcryptPasswordHasher,
        rounds=config.security.bcrypt_rounds,
    )

    audit_logger = providers.Singleton(
        FileAuditLogger,
        log_path=config.logging.audit_path,
    )

    # Services (factory - new instance per request)
    auth_service = providers.Factory(
        AuthService,
        user_repo=user_repo,
        hasher=password_hasher,
        audit=audit_logger,
    )

    order_service = providers.Factory(
        OrderService,
        user_repo=user_repo,
        payment=providers.Factory(StripePaymentProcessor, api_key=config.stripe.api_key),
        email=providers.Factory(SmtpEmailSender, host=config.email.smtp_host),
    )

Wiring to FastAPI

# main.py
from fastapi import FastAPI
from containers import AppContainer

def create_app() -> FastAPI:
    container = AppContainer()
    container.config.from_yaml("config.yaml")
    # or: container.config.from_dict({"database": {"url": "postgresql://..."}})

    app = FastAPI()
    container.wire(modules=[
        "routes.auth_routes",
        "routes.order_routes",
    ])

    app.container = container
    return app

app = create_app()

# routes/auth_routes.py
from fastapi import APIRouter, Depends
from dependency_injector.wiring import inject, Provide
from containers import AppContainer
from services.auth_service import AuthService

router = APIRouter()

@router.post("/login")
@inject
def login(
    email: str,
    password: str,
    auth_service: AuthService = Depends(Provide[AppContainer.auth_service]),
):
    user = auth_service.authenticate(email, password)
    if user is None:
        raise HTTPException(401, "Invalid credentials")
    return {"user_id": str(user.id)}

Provider Types

Provider	Behavior	Use Case
Singleton	Creates once, returns same instance	Database engine, config objects
Factory	Creates new instance each call	Services, repositories per request
ThreadSafeSingleton	Thread-safe singleton	Shared resources in threaded apps
Resource	Manages lifecycle (init + shutdown)	Connection pools, file handles
Configuration	Loads config from files/env/dicts	Application settings
Callable	Wraps a function call	Helper functions

Testing with Overrides

# tests/test_with_container.py
import pytest
from containers import AppContainer
from tests.fakes import FakeUserRepository, FakeHasher, FakeAuditLogger


@pytest.fixture
def container():
    c = AppContainer()
    c.user_repo.override(providers.Factory(FakeUserRepository))
    c.password_hasher.override(providers.Singleton(FakeHasher))
    c.audit_logger.override(providers.Singleton(FakeAuditLogger))
    yield c
    c.reset_singletons()
    c.user_repo.reset_override()
    c.password_hasher.reset_override()
    c.audit_logger.reset_override()


def test_login_with_fake_deps(container):
    auth = container.auth_service()
    # auth now uses FakeUserRepository, FakeHasher, FakeAuditLogger
    result = auth.authenticate("[email protected]", "password")
    assert result is None  # no users in fake repo

:::note Overrides Cascade When you override user_repo, every provider that depends on it (auth_service, order_service) automatically receives the fake. You do not need to override each consumer separately. :::

Part 5 - FastAPI's Depends System

FastAPI has its own DI mechanism built in. It is simpler than a full container but powerful enough for many applications.

Basic Usage

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

# Dependency: database session
def get_db() -> Session:
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

# Dependency: current user (depends on db)
def get_current_user(
    token: str = Header(...),
    db: Session = Depends(get_db),
) -> User:
    user_id = decode_jwt(token)
    user = db.query(UserModel).get(user_id)
    if not user:
        raise HTTPException(401, "Invalid token")
    return user

# Dependency: service (depends on db)
def get_order_service(db: Session = Depends(get_db)) -> OrderService:
    return OrderService(
        repo=SqlAlchemyOrderRepository(db),
        payment=StripePaymentProcessor(settings.stripe_key),
    )

# Route uses both
@app.post("/orders")
def create_order(
    items: list[dict],
    user: User = Depends(get_current_user),
    service: OrderService = Depends(get_order_service),
):
    return service.place_order(user.id, items)

Dependency Chains

FastAPI resolves dependency chains automatically. When two dependencies share a sub-dependency (like get_db), FastAPI caches the result within a single request - both get the same session.

Class-Based Dependencies

class Paginator:
    """Reusable dependency for pagination."""

    def __init__(self, max_limit: int = 100):
        self._max_limit = max_limit

    def __call__(
        self,
        offset: int = Query(0, ge=0),
        limit: int = Query(20, ge=1, le=100),
    ) -> dict:
        return {
            "offset": offset,
            "limit": min(limit, self._max_limit),
        }


paginate = Paginator(max_limit=50)


@app.get("/articles")
def list_articles(
    pagination: dict = Depends(paginate),
    service: ArticleService = Depends(get_article_service),
):
    return service.list(pagination["limit"], pagination["offset"])

Testing With dependency_overrides

# tests/test_api.py
from fastapi.testclient import TestClient

app = create_app()

# Override database dependency
def override_get_db():
    session = TestSessionLocal()
    try:
        yield session
    finally:
        session.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_create_order():
    response = client.post("/orders", json={"items": [{"id": 1, "qty": 2}]})
    assert response.status_code == 200

:::danger Avoid Circular Dependencies in FastAPI If get_service_a depends on get_service_b and get_service_b depends on get_service_a, FastAPI will raise a recursion error. Break cycles by introducing an intermediary or restructuring your service boundaries. :::

Part 6 - The DI Pyramid

Not every function needs dependency injection. Think of your codebase as a pyramid.

Layer	DI Usage	Example
Pure functions	None - these take data in, return data out	calculate_tax(price, rate), validate_email(email)
Services	Constructor injection - receive ports/adapters	OrderService(repo, payment, email)
Framework glue	Framework DI - Depends, containers, wiring	FastAPI routes, Click commands, Celery tasks

Where NOT to Use DI

# BAD: injecting a dependency for something that should be a pure function
class TaxCalculator:
    def __init__(self, rate_provider: TaxRateProvider):
        self._rate_provider = rate_provider

    def calculate(self, price: Decimal) -> Decimal:
        rate = self._rate_provider.get_rate()
        return price * rate

# GOOD: just pass the rate
def calculate_tax(price: Decimal, rate: Decimal) -> Decimal:
    return price * rate

If a function can be pure (takes all its inputs as arguments, returns a result, has no side effects), keep it pure. Do not wrap it in a class just to inject dependencies.

# BAD: DI for string formatting
class GreetingService:
    def __init__(self, formatter: StringFormatter):
        self._formatter = formatter

    def greet(self, name: str) -> str:
        return self._formatter.format(f"Hello, {name}!")

# GOOD: just format the string
def greet(name: str) -> str:
    return f"Hello, {name}!"

:::tip The Rule of Thumb Inject things that have side effects (database, network, file system) or that you might want to swap (hashing algorithm, payment provider). Do not inject things that are pure computation. :::

Part 7 - Advanced Patterns

Scoped Dependencies (Per-Request Lifetime)

from contextlib import contextmanager


class ScopedContainer:
    """Container that creates a scope per request."""

    def __init__(self, parent: Container) -> None:
        self._parent = parent
        self._scoped: dict[type, Any] = {}

    @contextmanager
    def scope(self):
        """Create a request scope - singletons within this scope."""
        scoped = {}
        try:
            yield ScopedResolver(self._parent, scoped)
        finally:
            # Cleanup scoped resources
            for obj in scoped.values():
                if hasattr(obj, "close"):
                    obj.close()


class ScopedResolver:
    def __init__(self, parent: Container, scoped: dict) -> None:
        self._parent = parent
        self._scoped = scoped

    def resolve(self, interface: type[T]) -> T:
        if interface in self._scoped:
            return self._scoped[interface]
        instance = self._parent.resolve(interface)
        self._scoped[interface] = instance
        return instance

Factory Injection (Creating Instances on Demand)

Sometimes a service needs to create multiple instances of a dependency during its lifetime, not just one at construction time.

from typing import Callable


class BatchProcessor:
    """Needs to create a new worker for each batch item."""

    def __init__(self, worker_factory: Callable[[], Worker]) -> None:
        self._create_worker = worker_factory

    def process_batch(self, items: list[dict]) -> list[Result]:
        results = []
        for item in items:
            worker = self._create_worker()  # new worker per item
            results.append(worker.process(item))
        return results


# Wiring
container.register(
    BatchProcessor,
    lambda c: BatchProcessor(
        worker_factory=lambda: c.resolve(Worker),
    ),
)

Decorator-Based Registration

# A registration decorator for simple cases
_registry: dict[type, type] = {}

def implements(interface: type):
    """Register a class as the implementation for an interface."""
    def decorator(cls: type) -> type:
        _registry[interface] = cls
        return cls
    return decorator


class OrderRepository(Protocol):
    def save(self, order: Order) -> None: ...


@implements(OrderRepository)
class PostgresOrderRepository:
    def __init__(self, session: Session) -> None:
        self._session = session

    def save(self, order: Order) -> None:
        self._session.merge(order_to_model(order))
        self._session.flush()


# Auto-discovery
def resolve(interface: type) -> type:
    if interface not in _registry:
        raise KeyError(f"No implementation registered for {interface}")
    return _registry[interface]

Part 8 - When DI Adds Value vs When It Is Overhead

DI Is Worth It When

Scenario	Why
External services (DB, API, email)	Swap real for fake in tests
Multiple implementations exist	Feature flags, A/B testing, migration
Team larger than 2-3 engineers	Clear contracts reduce integration friction
Long-lived production system	Adapters change; service logic stays
Complex object graphs	Container manages wiring automatically

DI Is Overhead When

Scenario	Better Alternative
Pure utility functions	Pass data as function arguments
Single implementation, never changes	Direct import is fine
Scripts and one-off tools	Keep it simple
Prototyping / hackathon	Inline everything, refactor later
Logging / metrics (cross-cutting)	Module-level logging.getLogger() is fine

The Pragmatic Approach

# Level 0: direct import (fine for simple cases)
import logging
logger = logging.getLogger(__name__)

# Level 1: function parameter (lightweight DI)
def process(data: dict, validator: Callable = default_validator):
    return validator(data)

# Level 2: constructor injection (standard DI)
class OrderService:
    def __init__(self, repo: OrderRepository, payment: PaymentGateway): ...

# Level 3: container (when Level 2 wiring gets unwieldy)
container = AppContainer()
container.config.from_yaml("config.yaml")

Advance to the next level only when the current one becomes painful.

:::danger Over-Injection If every class in your project receives 8+ constructor arguments, you likely have a design problem - not a DI problem. Consider breaking services into smaller, focused units or grouping related dependencies into aggregate objects. :::

Part 9 - Complete Example: FastAPI + Manual DI

A production-ready pattern that does not require any DI library.

# dependencies.py - the composition root for FastAPI
from functools import lru_cache
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from fastapi import Depends

from config import Settings
from adapters.repos import SqlUserRepo, SqlOrderRepo
from adapters.payment import StripeGateway
from adapters.email import SmtpSender
from services.auth import AuthService
from services.orders import OrderService


@lru_cache
def get_settings() -> Settings:
    return Settings()  # reads from env vars


@lru_cache
def get_engine(settings: Settings = Depends(get_settings)):
    return create_engine(settings.database_url)


def get_session(engine=Depends(get_engine)) -> Session:
    SessionLocal = sessionmaker(bind=engine)
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()


def get_auth_service(session: Session = Depends(get_session)) -> AuthService:
    return AuthService(
        user_repo=SqlUserRepo(session),
        hasher=BcryptHasher(),
        audit=DatabaseAuditLogger(session),
    )


def get_order_service(
    session: Session = Depends(get_session),
    settings: Settings = Depends(get_settings),
) -> OrderService:
    return OrderService(
        order_repo=SqlOrderRepo(session),
        user_repo=SqlUserRepo(session),
        payment=StripeGateway(settings.stripe_api_key),
        email=SmtpSender(settings.smtp_host, settings.smtp_port),
    )

# routes/orders.py
from fastapi import APIRouter, Depends
from dependencies import get_order_service, get_current_user
from services.orders import OrderService

router = APIRouter()

@router.post("/orders")
def create_order(
    items: list[dict],
    user: User = Depends(get_current_user),
    service: OrderService = Depends(get_order_service),
):
    return service.place_order(user.id, items)

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from main import app
from dependencies import get_session, get_order_service
from tests.fakes import FakeOrderRepo, FakePayment, FakeEmail


@pytest.fixture
def client():
    def _fake_order_service():
        return OrderService(
            order_repo=FakeOrderRepo(),
            user_repo=FakeUserRepo(),
            payment=FakePayment(),
            email=FakeEmail(),
        )

    app.dependency_overrides[get_order_service] = _fake_order_service
    yield TestClient(app)
    app.dependency_overrides.clear()

This pattern gives you full DI with zero extra libraries. The dependencies.py module is your composition root. dependency_overrides is your testing escape hatch.

Key Takeaways

• DI means providing dependencies from the outside instead of letting components create their own. The simplest form is passing arguments to __init__.
• Constructor injection is sufficient for most Python projects: you do not need a container until manual wiring becomes unwieldy (typically 20+ services).
• DI containers automate wiring: dependency-injector provides factories, singletons, configuration loading, and cascade overrides for testing.
• FastAPI's Depends is a lightweight DI system: it resolves dependency chains, caches within requests, and supports dependency_overrides for testing.
• Follow the DI pyramid: pure functions need no injection, services need constructor injection, and framework glue uses framework-specific DI.
• Inject side effects, not pure computation: if a function can take all its inputs as arguments and return a result with no external interaction, keep it as a simple function.

Graded Practice Challenges

Level 1 - Identify the Issue

Question 1: What is the DI anti-pattern in this code?

class ReportGenerator:
    def __init__(self):
        self._db = DatabaseConnection("postgresql://localhost/reports")

    def generate(self, report_id: int) -> str:
        data = self._db.query(f"SELECT * FROM reports WHERE id = {report_id}")
        return format_report(data)

Answer

Two issues: (1) ReportGenerator creates its own DatabaseConnection inside __init__, making it impossible to test without a real database. The connection should be injected. (2) The SQL query is constructed via string interpolation, which is a SQL injection vulnerability (not DI-related but worth noting).

Question 2: Why is this use of Depends problematic?

@app.post("/orders")
def create_order(
    items: list,
    service: OrderService = Depends(OrderService),
):
    return service.place_order(items)

Answer

Depends(OrderService) calls OrderService() with no arguments. If OrderService.__init__ requires dependencies (repository, payment gateway, etc.), this will fail at runtime. The correct approach is Depends(get_order_service) where get_order_service is a factory function that constructs OrderService with all its dependencies.

Question 3: When should you use a DI container instead of manual constructor injection?

Answer

Use a DI container when: (a) the number of services exceeds ~20 and the composition root becomes hard to maintain, (b) you need features like automatic lifecycle management (singletons, scoped), (c) you want configuration-driven wiring (load different implementations from config files), or (d) you need cascade overrides for testing. For smaller projects, manual constructor injection in a create_app() function is simpler and more explicit.

Level 2 - Refactoring Challenge

Refactor this tightly coupled notification system to use constructor injection:

class NotificationManager:
    def notify_user(self, user_id: int, message: str):
        db = psycopg2.connect("postgresql://localhost/app")
        cursor = db.cursor()
        cursor.execute("SELECT email, phone FROM users WHERE id = %s", (user_id,))
        row = cursor.fetchone()
        db.close()

        if row:
            email, phone = row
            import requests
            requests.post(
                "https://api.sendgrid.com/v3/mail/send",
                headers={"Authorization": "Bearer SG.xxx"},
                json={"to": email, "subject": "Alert", "text": message},
            )
            requests.post(
                "https://api.twilio.com/Messages",
                auth=("ACxxx", "token"),
                data={"To": phone, "Body": message},
            )

Produce: (a) a UserLookup port, (b) an EmailSender port, (c) an SmsSender port, (d) a refactored NotificationManager with constructor injection, (e) fakes for testing.

Level 3 - Design Challenge

Design the DI strategy for a multi-tenant SaaS application where:

• Each tenant has its own database schema
• Payment processing differs by tenant (Stripe vs PayPal vs manual billing)
• Email templates are tenant-specific
• Some tenants have custom business rules (implemented as plugins)

How would you structure the DI container to handle per-tenant resolution? Would you use scoped containers, factory injection, or another pattern? Produce the container definition and explain your reasoning.

What's Next

In the next lesson, Plugin Systems - Building Extensible Applications, we will explore how to build applications that third parties can extend without modifying your source code - using entry points, importlib.metadata, and runtime discovery.