# Composition Root

> Assembles dependency graphs at a single entry point using constructor injection, DI containers, and factory patterns to wire adapters to ports in hexagonal and layered architectures.

- Skill: `paulpas/composition-root` (Agent Skill)
- Install (CLI): `npx skillmds@latest add paulpas/composition-root`
- Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/composition-root/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- License: MIT
- Author: paulpas (https://skillmd.com/u/paulpas)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/paulpas/composition-root

---





# Composition Root & Dependency Injection Patterns

Acts as a senior software architect designing dependency injection composition roots. When loaded, the model assembles complete object graphs at a single entry point, selects appropriate DI strategies (manual wiring, factory functions, or container libraries), manages object lifecycles (singleton, per-request, transient), and produces concrete bootstrap code that enforces explicit dependency flow without service locator anti-patterns.

## TL;DR Checklist

- [ ] Identify every concrete class that must be instantiated and the interfaces it implements
- [ ] Build a single bootstrap function or module that creates all instances — never instantiate concretions inside business logic
- [ ] Prefer manual factory wiring for small-to-medium apps; reach for DI containers only when wiring complexity becomes unmanageable
- [ ] Use constructor injection exclusively — no property injection, no method injection, no defaults in constructors that hide dependencies
- [ ] Manage object lifecycles explicitly: singleton (shared state), per-request (HTTP scope), transient (new instance each use)
- [ ] Verify the composition root by calling it and exercising a complete use case with real infrastructure
- [ ] Ensure every test has its own bootstrap path that swaps production concretions for fakes

---

## When to Use

Use this skill when:

- Designing application startup code for a new service and needing to wire together core use cases, repository adapters, and external clients
- Refactoring an existing application where concrete classes are instantiated in scattered locations across the codebase (the "new everywhere" anti-pattern)
- Building both production and test bootstrap paths that share the same domain logic but swap different adapter implementations
- Migrating from hard-coded dependencies to explicit dependency injection without introducing a heavy DI framework
- Evaluating whether an application needs a DI container library or if manual wiring suffices based on object graph complexity

---

## When NOT to Use

Avoid this skill for:

- **Scripts and CLI tools with one-shot execution** — A script that reads config, processes data, and exits has no object graph worth assembling
- **Framework-managed applications using built-in DI** — If you're in FastAPI (with `Depends`), Django (with service classes), or similar frameworks that provide their own wiring mechanisms, use the framework's patterns rather than introducing a separate composition root
- **Prototypes and hackathon projects** — The goal is speed; explicit dependency graphs add ceremony that delays delivery
- **When every dependency is truly singleton with no lifecycle concerns** — If you only need one shared instance of each class with no configuration differences between environments, direct imports in a service registry may suffice

---

## Core Workflow

1. **Inventory all concrete dependencies** — List every infrastructure adapter, external client, utility service, and domain service that your application needs at runtime. For each dependency, note: the interface it implements (Protocol or ABC), whether it holds mutable state requiring shared instances, and which other services depend on it.
   **Checkpoint:** Draw a directed graph where nodes are concrete classes and edges represent constructor dependencies. The composition root is the entry point that creates every node in topological order.

2. **Choose a wiring strategy** — Select between three approaches based on object graph complexity: (a) manual factory function — one function returning a fully wired service, suitable for up to ~15 concrete classes; (b) DI container with explicit registration — use `dependency-injector` or a simple dict-based registry for larger apps; (c) per-request scope wiring — required when services have request-scoped state (e.g., HTTP request context). For most applications, manual factory functions are sufficient and preferred.
   **Checkpoint:** If the wiring function exceeds 50 lines of sequential `container.register(...)` calls or nested constructor arguments exceed 5 parameters, consider a DI container library.

3. **Implement the factory function** — Write a single function (commonly named `build_application`, `create_app`, or `bootstrap`) that: imports all concrete classes, instantiates leaf dependencies first (those with no injected dependencies), then builds upward through the dependency graph. Each instantiation passes explicitly constructed arguments — never `None` placeholders or environment-variable lookups inside business logic.
   **Checkpoint:** The factory function's parameter list should accept only configuration values (connection strings, API keys, feature flags). Every internal dependency must be created within the function body.

4. **Create test bootstrap paths** — For each production factory function, create a corresponding test version that substitutes every injected dependency with a fake or mock implementation. Use the same interface types so the calling code is identical between production and test — only the bootstrap differs.
   **Checkpoint:** Call the test bootstrap function and run a complete end-to-end unit test (one use case from external input to domain output). It must execute without any network calls, database connections, or filesystem access.

5. **Wire per-request dependencies** — If your application serves HTTP requests, create a second factory function that runs once per request. This function receives the request context (e.g., `Request` object from FastAPI) and creates request-scoped services like database sessions, current-user resolvers, and request-specific caches. The production root injects this request factory into the relevant adapters.
   **Checkpoint:** Request-scoped services must not leak state across requests — each invocation of the per-request factory creates fresh instances with no shared mutable state from previous requests.

---

## Implementation Patterns

### Pattern 1: Manual Composition Root (Factory Function Wiring)

The simplest and most explicit wiring approach. A single factory function imports all concrete classes, constructs leaf dependencies first, then builds upward through the dependency graph. Ideal for applications up to ~15 concrete classes.

```python
from __future__ import annotations

# --- Domain layer: Protocols (interfaces) ---
from typing import Protocol


class OrderRepository(Protocol):
    """Interface that order service needs from persistence."""

    def save(self, order_id: str, user_id: str, total_cents: int) -> None: ...
    def get(self, order_id: str) -> dict | None: ...


class NotificationChannel(Protocol):
    """Interface for sending notifications to users."""

    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool: ...


class Logger(Protocol):
    """Interface for logging infrastructure calls."""

    def info(self, message: str) -> None: ...
    def error(self, message: str) -> None: ...


# --- Infrastructure layer: concrete implementations ---
class PostgresOrderRepo:
    """Persistence adapter backed by PostgreSQL."""

    def __init__(self, connection_string: str, logger: Logger) -> None:
        self._logger = logger
        # Real implementation would create a real database connection here

    def save(self, order_id: str, user_id: str, total_cents: int) -> None:
        self._logger.info(f"Saving order {order_id} for user {user_id}")
        # Real code: execute INSERT via psycopg2 or SQLAlchemy

    def get(self, order_id: str) -> dict | None:
        self._logger.info(f"Fetching order {order_id}")
        # Real code: execute SELECT via psycopg2 or SQLAlchemy
        return {"id": order_id, "user_id": user_id, "total_cents": total_cents}


class EmailNotificationChannel:
    """Sends order confirmations via SMTP email."""

    def __init__(self, smtp_host: str, smtp_port: int, sender: str, logger: Logger) -> None:
        self._smtp_host = smtp_host
        self._smtp_port = smtp_port
        self._sender = sender
        self._logger = logger

    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool:
        self._logger.info(f"Sending confirmation to {user_email} for order {order_id}")
        # Real code: connect to SMTP server and send email
        return True


class ConsoleLogger:
    """Simple logger writing to stdout/stderr."""

    def info(self, message: str) -> None:
        print(f"[INFO] {message}")

    def error(self, message: str) -> None:
        print(f"[ERROR] {message}", flush=True)


# --- Domain layer: services (depend on protocols only) ---
class OrderService:
    """Core business logic for order management."""

    def __init__(self, repo: OrderRepository, notifier: NotificationChannel, logger: Logger) -> None:
        self._repo = repo
        self._notifier = notifier
        self._logger = logger

    def create_order(self, user_id: str, user_email: str, items: list[dict], total_cents: int) -> str:
        """Create a new order with the given items."""
        if not items:
            raise ValueError("Order must contain at least one item")

        order_id = f"ord_{user_id}_{len(items)}"  # Simplified ID generation

        self._repo.save(order_id=order_id, user_id=user_id, total_cents=total_cents)
        self._notifier.send_order_confirmation(user_email, order_id, total_cents)
        self._logger.info(f"Order {order_id} created for user {user_id}")

        return order_id


class ProductService:
    """Core business logic for product catalog."""

    def __init__(self, repo: OrderRepository, logger: Logger) -> None:  # Using same repo as example
        self._repo = repo
        self._logger = logger


# --- Composition Root: single entry point for dependency wiring ---
def build_production_app(
    db_connection_string: str = "postgresql://localhost/orders",
    smtp_host: str = "smtp.gmail.com",
    smtp_port: int = 587,
    email_sender: str = "noreply@example.com",
) -> OrderService:
    """Assemble the complete application object graph for production.

    This is the ONLY place in the application where concrete classes are
    instantiated and imported together. All business logic depends only
    on Protocol interfaces injected through constructors.
    """
    # 1. Leaf dependencies (no injected dependencies of their own)
    logger: Logger = ConsoleLogger()

    # 2. Infrastructure adapters (depend on leaf dependencies)
    repo: OrderRepository = PostgresOrderRepo(connection_string=db_connection_string, logger=logger)
    notifier: NotificationChannel = EmailNotificationChannel(
        smtp_host=smtp_host,
        smtp_port=smtp_port,
        sender=email_sender,
        logger=logger,
    )

    # 3. Domain services (depend on infrastructure adapters via protocols)
    order_service: OrderService = OrderService(repo=repo, notifier=notifier, logger=logger)

    # 4. Return the top-level service — the composition root returns
    #    the entry point of the application, not every individual service
    return order_service


# Usage (in main.py or entry point):
# app = build_production_app(db_connection_string="postgresql://prod-db/orders")
# order_id = app.create_order(user_id="u123", user_email="alice@example.com", items=[{"sku": "A1", "qty": 2}], total_cents=4998)
```

### Pattern 2: DI Container with Explicit Registration (dependency-injector Library)

For applications where manual wiring becomes unwieldy, use a lightweight DI container. The `dependency-injector` library provides explicit registration, lifecycle management, and per-request scoping without magic or hidden dependencies.

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


# --- Protocol definitions (abbreviated — same as Pattern 1) ---
class OrderRepository(Protocol):
    def save(self, order_id: str, user_id: str, total_cents: int) -> None: ...
    def get(self, order_id: str) -> dict | None: ...


class NotificationChannel(Protocol):
    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool: ...


class Logger(Protocol):
    def info(self, message: str) -> None: ...
    def error(self, message: str) -> None: ...


# --- Concrete implementations (abbreviated) ---
class PostgresOrderRepo:
    def __init__(self, connection_string: str, logger: Logger) -> None:
        pass  # Infrastructure details omitted for brevity

    def save(self, order_id: str, user_id: str, total_cents: int) -> None: ...
    def get(self, order_id: str) -> dict | None: ...


class EmailNotificationChannel:
    def __init__(self, smtp_host: str, smtp_port: int, sender: str, logger: Logger) -> None:
        pass  # Infrastructure details omitted

    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool:
        return True


class ConsoleLogger:
    def info(self, message: str) -> None: ...
    def error(self, message: str) -> None: ...


class OrderService:
    @inject
    def __init__(
        self,
        repo: OrderRepository = Provide["order_repo"],
        notifier: NotificationChannel = Provide["notification_channel"],
        logger: Logger = Provide["logger"],
    ) -> None:
        # Constructor uses dependency-injector wiring annotations
        # For manual testing, pass explicitly:
        #   OrderService(repo=..., notifier=..., logger=...)
        self._repo = repo
        self._notifier = notifier
        self._logger = logger

    def create_order(self, user_id: str, user_email: str, items: list[dict], total_cents: int) -> str:
        if not items:
            raise ValueError("Order must contain at least one item")
        order_id = f"ord_{user_id}_{len(items)}"
        self._repo.save(order_id=order_id, user_id=user_id, total_cents=total_cents)
        self._notifier.send_order_confirmation(user_email, order_id, total_cents)
        return order_id


# --- Container Definition: declarative wiring ---
class AppContainer(containers.DeclarativeContainer):
    """DI container defining the application's dependency graph."""

    # Configuration (external inputs)
    config = providers.Configuration()

    # Leaf dependencies
    logger = providers.Factory(
        ConsoleLogger,
    )

    # Infrastructure adapters
    order_repo = providers.Factory(
        PostgresOrderRepo,
        connection_string=config.db.connection_string,
        logger=logger,
    )

    notification_channel = providers.Factory(
        EmailNotificationChannel,
        smtp_host=config.smtp.host,
        smtp_port=config.smtp.port,
        sender=config.smtp.sender,
        logger=logger,
    )

    # Domain services
    order_service = providers.Factory(
        OrderService,
    )


# --- Usage with container ---
def main() -> None:
    """Bootstrap application using the DI container."""
    container = AppContainer()

    # Override configuration from environment variables in production
    container.config.db.connection_string.from_env("DATABASE_URL")
    container.config.smtp.host.from_env("SMTP_HOST", default="smtp.gmail.com")
    container.config.smtp.port.from_env("SMTP_PORT", as_=int, default=587)
    container.config.smtp.sender.from_env("EMAIL_SENDER", default="noreply@example.com")

    # Wire into FastAPI or Flask middleware for injection support
    # order_service = container.order_service()  # Resolved automatically

    # Or resolve manually (no framework integration needed):
    order_service = container.order_service()
    order_id = order_service.create_order(
        user_id="u123",
        user_email="alice@example.com",
        items=[{"sku": "A1", "qty": 2}],
        total_cents=4998,
    )
    print(f"Created order: {order_id}")

    # Clean shutdown
    container.shutdown()


# --- Test container: swap dependencies for testing ---
class TestContainer(containers.DeclarativeContainer):
    """Test-specific container with fakes injected."""

    config = providers.Configuration()

    logger = providers.Singleton(  # Shared singleton across tests
        ConsoleLogger(),
    )

    order_repo = providers.Factory(
        FakeOrderRepo,  # Hand-written fake implementing OrderRepository
    )

    notification_channel = providers.Factory(
        FakeNotificationChannel,  # Hand-written fake
    )

    order_service = providers.Factory(OrderService)


def run_integration_test() -> None:
    """Demonstrate test wiring with fakes."""
    container = TestContainer()
    try:
        service = container.order_service()
        result = service.create_order(
            user_id="test_user",
            user_email="test@example.com",
            items=[{"sku": "X1", "qty": 1}],
            total_cents=999,
        )
        assert result is not None
    finally:
        container.shutdown()


# --- Fakes for testing (abbreviated) ---
class FakeOrderRepo:
    def __init__(self) -> None:
        self._orders: dict[str, dict] = {}

    def save(self, order_id: str, user_id: str, total_cents: int) -> None:
        self._orders[order_id] = {"user_id": user_id, "total_cents": total_cents}

    def get(self, order_id: str) -> dict | None:
        return self._orders.get(order_id)


class FakeNotificationChannel:
    def __init__(self) -> None:
        self.sent_messages: list[dict] = []

    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool:
        self.sent_messages.append({"to": user_email, "order_id": order_id})
        return True
```

### Pattern 3: Per-Request Scope Wiring (HTTP Request Lifecycle)

In web applications, certain services must be scoped to individual HTTP requests — database sessions, request context data, per-user caches. This pattern creates a factory that runs once per request, ensuring no state leaks between concurrent requests.

```python
from typing import Callable


class DbSession:
    """Represents a database session scoped to a single HTTP request."""

    def __init__(self, connection_string: str) -> None:
        self._connection_string = connection_string
        # Real code: create psycopg2 connection or SQLAlchemy session

    def execute(self, query: str, params: tuple = ()) -> list[dict]:
        """Execute a database query within this request's scope."""
        # Real code: return self.session.execute(query, params).fetchall()
        return []


class RequestContext:
    """Holds data specific to the current HTTP request."""

    def __init__(self, user_id: str | None = None, request_headers: dict[str, str] | None = None) -> None:
        self.user_id = user_id
        self.request_headers = request_headers or {}


class RequestScopedOrderRepo:
    """Order repository that uses a per-request database session."""

    def __init__(self, db_session_factory: Callable[[], DbSession], logger: Logger) -> None:
        self._db_session_factory = db_session_factory
        self._logger = logger

    def get_by_id(self, order_id: str) -> dict | None:
        session = self._db_session_factory()  # Gets the request-scoped session
        results = session.execute("SELECT * FROM orders WHERE id = %s", (order_id,))
        return results[0] if results else None

    def save(self, order_id: str, user_id: str, total_cents: int) -> None:
        session = self._db_session_factory()  # Same request-scoped session
        self._logger.info(f"Saving order {order_id}")
        session.execute(
            "INSERT INTO orders (id, user_id, total_cents) VALUES (%s, %s, %s)",
            (order_id, user_id, total_cents),
        )


class RequestScopedNotificationChannel:
    """Notification channel that tracks sent messages per request."""

    def __init__(self, smtp_host: str, context_getter: Callable[[], RequestContext]) -> None:
        self._smtp_host = smtp_host
        self._context_getter = context_getter

    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool:
        ctx = self._context_getter()  # Get current request's context
        # Send email with request-scoped metadata (trace IDs, etc.)
        return True


# --- Composition root with per-request factory ---
class HttpRequestFactory:
    """Creates request-scoped services for each incoming HTTP request.

    This runs ONCE PER REQUEST in middleware or framework handler.
    It ensures database sessions and contexts are properly scoped.
    """

    def __init__(self, db_connection_string: str) -> None:
        self._db_connection_string = db_connection_string

    def create_scope(self, user_id: str | None = None) -> dict:
        """Create a new request scope with all scoped services.

        Returns a dict of scoped services that middleware injects
        into the request context for downstream handlers to use.
        """
        # Request-scoped dependencies created fresh per request
        db_session = DbSession(self._db_connection_string)
        request_context = RequestContext(user_id=user_id)

        # Loggers are typically shared (singleton), not request-scoped
        logger: Logger = ConsoleLogger()

        # Build scoped adapters
        order_repo = RequestScopedOrderRepo(
            db_session_factory=lambda: db_session,
            logger=logger,
        )
        notifier = RequestScopedNotificationChannel(
            smtp_host="smtp.gmail.com",
            context_getter=lambda: request_context,
        )

        return {
            "db_session": db_session,
            "order_repo": order_repo,
            "notification_channel": notifier,
            "request_context": request_context,
        }


# --- Middleware integration example (conceptual) ---
def setup_request_middleware(app):  # type: ignore[no-untyped-def]
    """Attach request-scoped factory as middleware for FastAPI/Flask."""
    http_factory = HttpRequestFactory(
        db_connection_string="postgresql://localhost/orders",
    )

    @app.middleware("http")  # type: ignore[misc, unused-ignore]
    async def request_scope_middleware(request, call_next):  # type: ignore[no-untyped-def]
        user_id = request.headers.get("x-user-id")  # type: ignore[union-attr]
        scope = http_factory.create_scope(user_id=user_id)

        # Inject scoped services into the request (framework-specific)
        # request.state.db_session = scope["db_session"]
        # request.state.order_repo = scope["order_repo"]

        response = await call_next(request)  # type: ignore[no-untyped-call]

        # Cleanup: close database session at end of request
        scope["db_session"].execute("ROLLBACK")  # type: ignore[union-attr]

        return response
```

### Pattern 4: Configuration-Based Assembly (Loading Adapter Selection from Config)

Real applications need to select different adapters based on environment — SQLite for development, PostgreSQL for production. This pattern reads configuration and instantiates the appropriate concrete class while maintaining the same interface contract.

```python
from __future__ import annotations
import os


# --- Protocol definitions ---
class OrderRepository(Protocol):
    def save(self, order_id: str, user_id: str, total_cents: int) -> None: ...
    def get(self, order_id: str) -> dict | None: ...


class NotificationChannel(Protocol):
    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool: ...


class Logger(Protocol):
    def info(self, message: str) -> None: ...
    def error(self, message: str) -> None: ...


# --- Multiple implementations for the same protocol ---
class PostgresOrderRepo:
    def __init__(self, connection_string: str, logger: Logger) -> None: ...
    def save(self, order_id: str, user_id: str, total_cents: int) -> None: ...
    def get(self, order_id: str) -> dict | None: ...


class SQLiteOrderRepo:
    """Development/test adapter backed by SQLite for zero-configuration local runs."""

    def __init__(self, db_path: str = "orders.db", logger: Logger | None = None) -> None:
        self._db_path = db_path
        self._logger = logger or ConsoleLogger()

    def save(self, order_id: str, user_id: str, total_cents: int) -> None:
        import sqlite3
        with sqlite3.connect(self._db_path) as conn:
            conn.execute(
                "INSERT OR REPLACE INTO orders (id, user_id, total_cents) VALUES (?, ?, ?)",
                (order_id, user_id, total_cents),
            )

    def get(self, order_id: str) -> dict | None:
        import sqlite3
        with sqlite3.connect(self._db_path) as conn:
            row = conn.execute("SELECT id, user_id, total_cents FROM orders WHERE id = ?", (order_id,)).fetchone()
            return {"id": row[0], "user_id": row[1], "total_cents": row[2]} if row else None


class SmtpNotificationChannel:
    """Sends emails via SMTP — used in production and staging."""

    def __init__(self, host: str, port: int = 587, logger: Logger | None = None) -> None:
        self._host = host
        self._port = port
        self._logger = logger or ConsoleLogger()

    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool:
        self._logger.info(f"Sending email to {user_email}")
        return True


class FakeNotificationChannel:
    """No-op notification channel used in testing and CI."""

    def __init__(self) -> None:
        self._sent: list[dict] = []

    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool:
        self._sent.append({"to": user_email, "order_id": order_id})
        return True


# --- Configuration-based factory resolver ---
def resolve_repo_factory(config: dict) -> tuple[OrderRepository, Logger]:
    """Select and construct the appropriate repository based on environment config.

    This is where environment-specific adapter selection happens — the composition
    root delegates to a resolver that reads config and returns concrete instances.
    """
    logger = ConsoleLogger()
    env = config.get("environment", "development")

    if env == "production":
        repo: OrderRepository = PostgresOrderRepo(
            connection_string=config["database"]["connection_string"],
            logger=logger,
        )
    elif env == "testing":
        # Use SQLite for testing — zero external dependencies
        repo = SQLiteOrderRepo(db_path=":memory:", logger=logger)
    else:
        # Default to SQLite for local development
        repo = SQLiteOrderRepo(db_path=config.get("database", {}).get("path", "orders.db"), logger=logger)

    return repo, logger


def resolve_notifier_factory(config: dict) -> tuple[NotificationChannel, Logger]:
    """Select and construct the appropriate notification channel based on environment config."""
    logger = ConsoleLogger()
    env = config.get("environment", "development")

    if env == "testing":
        notifier: NotificationChannel = FakeNotificationChannel()
    elif env in ("production", "staging"):
        notifier = SmtpNotificationChannel(
            host=config["smtp"]["host"],
            port=config["smtp"].get("port", 587),
            logger=logger,
        )
    else:
        # Development: log notifications instead of sending real emails
        notifier = FakeNotificationChannel()

    return notifier, logger


# --- Application bootstrap with configuration-based assembly ---
def build_app(config: dict | None = None) -> OrderService:  # type: ignore[name-defined]
    """Build the application using environment-specific adapter selection.

    The config dict is read from environment variables or a YAML/JSON file
    and determines which concrete adapters are instantiated for each port.
    """
    if config is None:
        config = {
            "environment": os.getenv("APP_ENV", "development"),
            "database": {
                "connection_string": os.getenv("DATABASE_URL", ""),
                "path": "orders.db",
            },
            "smtp": {
                "host": os.getenv("SMTP_HOST", "localhost"),
                "port": int(os.getenv("SMTP_PORT", "1025")),
            },
        }

    repo, logger = resolve_repo_factory(config)
    notifier, _ = resolve_notifier_factory(config)

    # Ensure OrderService is imported or defined above
    return OrderService(repo=repo, notifier=notifier, logger=logger)


# --- Environment-specific usage ---
def main() -> None:
    """Entry point — configuration comes from environment variables."""
    app = build_app()
    order_id = app.create_order(
        user_id="user_001",
        user_email="alice@example.com",
        items=[{"sku": "WIDGET-A", "qty": 3}],
        total_cents=2997,
    )
    print(f"Order created: {order_id}")


# --- Environment configuration examples ---
PRODUCTION_CONFIG = {
    "environment": "production",
    "database": {"connection_string": "postgresql://prod-server/orders"},
    "smtp": {"host": "smtp.gmail.com", "port": 587},
}

DEVELOPMENT_CONFIG = {
    "environment": "development",
    "database": {"path": "dev_orders.db"},
    "smtp": {"host": "localhost", "port": 1025},
}

TESTING_CONFIG = {
    "environment": "testing",
    "database": {"connection_string": ""},
    "smtp": {"host": "", "port": 0},
}
```

### Pattern 5: Composition Root Testing (Verifying the Dependency Graph)

A composition root is only as good as its testability. This pattern demonstrates how to verify that every dependency in your graph resolves correctly and that test fakes produce deterministic results.

```python
from __future__ import annotations


class FakeOrderRepo:
    """Complete fake implementing OrderRepository for testing."""

    def __init__(self) -> None:
        self._orders: dict[str, dict] = {}
        self.save_calls: list[tuple[str, str, int]] = []
        self.get_calls: list[str] = []

    def save(self, order_id: str, user_id: str, total_cents: int) -> None:
        self.save_calls.append((order_id, user_id, total_cents))
        self._orders[order_id] = {"id": order_id, "user_id": user_id, "total_cents": total_cents}

    def get(self, order_id: str) -> dict | None:
        self.get_calls.append(order_id)
        return self._orders.get(order_id)


class FakeNotificationChannel:
    """Complete fake implementing NotificationChannel for testing."""

    def __init__(self) -> None:
        self.sent_messages: list[dict] = []
        self.send_count: int = 0

    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool:
        self.sent_messages.append({
            "to": user_email,
            "order_id": order_id,
            "total_cents": total_cents,
        })
        self.send_count += 1
        return True


class FakeLogger:
    """Complete fake implementing Logger for testing."""

    def __init__(self) -> None:
        self.info_messages: list[str] = []
        self.error_messages: list[str] = []

    def info(self, message: str) -> None:
        self.info_messages.append(message)

    def error(self, message: str) -> None:
        self.error_messages.append(message)


def build_test_app() -> OrderService:  # type: ignore[name-defined]
    """Test bootstrap — all dependencies are fakes, zero infrastructure.

    This function mirrors the production build_app() but substitutes every
    injected dependency with a Fake implementation. The calling code is
    identical between test and production.
    """
    repo = FakeOrderRepo()
    notifier = FakeNotificationChannel()
    logger = FakeLogger()

    return OrderService(repo=repo, notifier=notifier, logger=logger)


# --- Test examples demonstrating composition root verification ---

def test_order_creation_completes_full_workflow() -> None:
    """Verify that the entire dependency graph resolves and a use case executes.

    This is the smoke test for the composition root: if this fails, something
    in the wiring chain is broken — missing dependency, wrong type, or circular reference.
    """
    app = build_test_app()  # Creates full object graph with fakes

    result_order_id = app.create_order(
        user_id="test_user",
        user_email="tester@example.com",
        items=[{"sku": "A1", "qty": 2}],
        total_cents=4998,
    )

    # Verify persistence layer was called correctly
    assert result_order_id is not None
    assert len(app._repo.save_calls) == 1
    saved = app._repo.save_calls[0]
    assert saved[1] == "test_user"  # user_id
    assert saved[2] == 4998  # total_cents

    # Verify notification was sent
    assert app._notifier.send_count == 1
    assert app._notifier.sent_messages[0]["to"] == "tester@example.com"


def test_invalid_order_rejected_before_persistence() -> None:
    """Verify business logic validation runs before any adapter is called.

    This confirms that the dependency graph wiring doesn't bypass domain rules —
    an empty-order error should be raised without touching the repo or notifier.
    """
    app = build_test_app()

    try:
        app.create_order(
            user_id="test_user",
            user_email="tester@example.com",
            items=[],  # Empty items list — validation should fail
            total_cents=0,
        )
    except ValueError as exc:
        assert "at least one item" in str(exc)
        # Neither persistence nor notification should have been called
        assert app._repo.save_calls == []
        assert app._notifier.send_count == 0


def test_composition_root_resolves_all_dependencies() -> None:
    """Verify the composition root creates a complete, resolvable dependency graph.

    This test ensures no dependency is left unresolved or None at runtime.
    Run this after every structural change to the wiring code.
    """
    app = build_test_app()

    # Every injected dependency must be a concrete instance, not None
    assert isinstance(app._repo, FakeOrderRepo), "Repository was not wired"
    assert isinstance(app._notifier, FakeNotificationChannel), "Notifier was not wired"
    assert isinstance(app._logger, FakeLogger), "Logger was not wired"

    # Each dependency must have its expected interface
    assert hasattr(app._repo, "save"), "Repo missing 'save' method"
    assert hasattr(app._repo, "get"), "Repo missing 'get' method"
    assert hasattr(app._notifier, "send_order_confirmation"), "Notifier missing 'send_order_confirmation'"


def test_graph_isolation_between_test_cases() -> None:
    """Verify that each test case gets a fresh composition root.

    Fakes must not leak state between tests — calling build_test_app()
    multiple times should produce independent object graphs.
    """
    app_1 = build_test_app()
    app_1.create_order("user_a", "a@example.com", [{"sku": "X"}], 100)

    app_2 = build_test_app()  # New graph — must be independent of app_1
    app_2.create_order("user_b", "b@example.com", [{"sku": "Y"}], 200)

    assert len(app_1._repo.save_calls) == 1
    assert len(app_2._repo.save_calls) == 1
    assert app_1._repo.get("user_b") is None  # user_b should not exist in app_1's graph
    assert app_2._repo.get("user_a") is None  # user_a should not exist in app_2's graph

```

## Anti-Patterns

### Anti-Pattern 1: Service Locator (Hidden Dependencies)

The service locator pattern conceals dependencies by allowing any module to query a global registry for its dependencies. This makes the dependency graph invisible to type checkers, reviewers, and static analysis tools.

```python
# ❌ BAD — Service Locator: dependencies hidden behind a global registry.
# No way to tell what OrderService needs without reading its source code.
# Cannot inject fakes in tests without monkey-patching the global container.


class ServiceLocator:
    """Global dependency registry — an anti-pattern."""

    _registry: dict[str, object] = {}

    @classmethod
    def register(cls, name: str, instance: object) -> None:
        cls._registry[name] = instance

    @classmethod
    def get(cls, name: str) -> object:
        if name not in cls._registry:
            raise LookupError(f"Service '{name}' not found in locator")
        return cls._registry[name]


class BadOrderService:
    """OrderService that discovers its dependencies from the global locator."""

    def create_order(self, user_id: str, user_email: str, items: list[dict], total_cents: int) -> str:
        if not items:
            raise ValueError("Items required")

        # ❌ Dependency discovered at runtime via global lookup — invisible to static analysis
        repo = ServiceLocator.get("order_repo")  # Hidden dependency!
        notifier = ServiceLocator.get("notification_channel")  # Hidden dependency!
        logger = ServiceLocator.get("logger")  # Hidden dependency!

        order_id = f"ord_{user_id}"
        repo.save(order_id, user_id, total_cents)  # Type: Any — no static checking!
        notifier.send_order_confirmation(user_email, order_id, total_cents)
        logger.info(f"Order created")
        return order_id


# ✅ GOOD — Constructor injection: all dependencies are explicit and type-checked.

from typing import Protocol


class GoodRepo(Protocol):
    def save(self, order_id: str, user_id: str, total_cents: int) -> None: ...


class GoodNotifier(Protocol):
    def send_order_confirmation(self, user_email: str, order_id: str, total_cents: int) -> bool: ...


class GoodLogger(Protocol):
    def info(self, message: str) -> None: ...


class GoodOrderService:
    """OrderService with explicit constructor injection — all dependencies visible."""

    def __init__(self, repo: GoodRepo, notifier: GoodNotifier, logger: GoodLogger) -> None:
        self._repo = repo
        self._notifier = notifier
        self._logger = logger

    def create_order(self, user_id: str, user_email: str, items: list[dict], total_cents: int) -> str:
        if not items:
            raise ValueError("Items required")

        order_id = f"ord_{user_id}"
        self._repo.save(order_id, user_id, total_cents)  # Type-safe — mypy knows the interface
        self._notifier.send_order_confirmation(user_email, order_id, total_cents)
        self._logger.info(f"Order created")
        return order_id
```

### Anti-Pattern 2: New Everywhere (Scattered Instantiation)

When concrete classes are instantiated throughout the codebase — in constructors, method bodies, and module-level globals — there is no single place to swap implementations. This makes testing impossible without mocking infrastructure.

```python
# ❌ BAD — Concrete instances created at every level of the hierarchy.
# Changing from SQLite to PostgreSQL requires editing dozens of files.


class BadOrderService:
    """Creates its own database connection — tightly coupled and untestable."""

    def __init__(self) -> None:
        # Each service creates its own infrastructure dependency
        import sqlite3
        self._db = sqlite3.connect("orders.db")  # Hardcoded path, hardcoded library

    def create_order(self, user_id: str, items: list[dict]) -> str:
        order_id = f"ord_{user_id}"
        cursor = self._db.cursor()
        cursor.execute(
            "INSERT INTO orders (id, user_id) VALUES (?, ?)",
            (order_id, user_id),
        )
        self._db.commit()
        return order_id


class BadNotificationService:
    """Creates its own HTTP client — cannot be replaced for testing."""

    def __init__(self) -> None:
        import httpx
        # ❌ Concrete class instantiation inside the service constructor
        self._client = httpx.Client(base_url="https://api.example.com")  # Hidden dependency!

    def send(self, email: str, subject: str, body: str) -> bool:
        response = self._client.post("/send", json={"to": email, "subject": subject, "body": body})
        return response.status_code == 200


# ✅ GOOD — A single composition root creates all infrastructure and injects it.

def build_good_app() -> dict[str, object]:
    """One function that assembles the entire depe

…(truncated)