FastAPI Dependency Injection : Why Dishka is Superior

Two Dependency Injection Philosophies: FastAPI Depends() vs. Dishka IoC

– FastAPI: Request-Centric Parameter Parsing
FastAPI ships with a built-in dependency injection system based on the Depends() callable. Designed primarily as an HTTP request-parsing utility, it couples parameter extraction (headers, query params, security tokens) with service instantiation directly inside route definitions. It prioritizes out-of-the-box convenience and OpenAPI schema generation over strict architectural separation.

– Dishka: Pure Inversion of Control & Domain Decoupling
Dishka operates as a standalone, framework-agnostic IoC container designed to isolate core domain logic from transport protocols. It enforces constructor injection on pure POJO services, uses type reflection for zero-boilerplate auto-wiring, and manages hierarchical lifecycles (scopes) through a deterministic Directed Acyclic Graph (DAG) fully validated at startup.

Architecture Comparison & Engineering Outcomes

Architectural Criterion FastAPI Depends() Dedicated IoC Container (Dishka)
Injection Target HTTP endpoint function signatures only Class constructors (__init__) of pure POJOs
Graph Resolution Manual chaining of nested factory functions Automated type inspection (Auto-wiring)
Graph Validation ❌ Runtime only (when the route is first requested) ✅ Boot time (Deterministic fail-fast at startup)
Test Isolation ❌ Shared mutable dict (app.dependency_overrides) ✅ Hermetic container instances per test case
Scope Granularity Binary: Global singleton or per-request generator Multi-tier: APP, REQUEST, ACTION, STEP
Framework Portability ❌ Coupled to Starlette HTTP request lifecycles ✅ Universal across FastAPI, CLI, FastStream, Taskiq
Software Impact FastAPI Depends() Dedicated IoC Container (Dishka)
Codebase Verbosity High: Exponential boilerplate due to intermediate factories Minimal: Declarative single-line provider registrations
Runtime Fragility High: Hidden dependency bugs detonate in production on cold routes Zero: Total DAG integrity enforced prior to accepting traffic
Test Maintainability Fragile: High risk of test pollution and race conditions in parallel CI Robust: Thread-safe container overrides scoped strictly per test
Architectural Purity Leaky Abstraction: Business logic coupled to the web framework; dependencies tangled with route parameters Clean Architecture: Pure POJO domain completely protocol-agnostic
Overall Verdict Suited for quick prototypes & simple CRUD endpoints Industrial standard for robust, scalable production systems

Part 1: The 6 Concrete Limitations of FastAPI Depends()

1. Signature Pollution & Direct Route Coupling

FastAPI cannot resolve dependencies through class constructors on its own. By default, every individual endpoint must explicitly declare every single infrastructure dependency in its parameter signature. Adding a new route or extending a service requires copy-pasting lines of technical plumbing across all handler functions.

from typing import Annotated
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
 
router = APIRouter()
 
# Every route signature is polluted with identical technical plumbing
@router.post("/orders/{order_id}/process")
async def process_order(
    order_id: int,
    db: Annotated[AsyncSession, Depends(get_db)],
    order_repo: Annotated[OrderRepository, Depends(get_order_repo)],
    payment_gateway: Annotated[PaymentGateway, Depends(get_payment_gateway)],
    email_notifier: Annotated[EmailNotifier, Depends(get_email_notifier)],
    audit_logger: Annotated[AuditLogger, Depends(get_audit_logger)],
) -> dict[str, str]:
    return {"status": "processed"}
 
# Duplicated boilerplate across every additional endpoint of the same aggregate
@router.post("/orders/{order_id}/cancel")
async def cancel_order(
    order_id: int,
    db: Annotated[AsyncSession, Depends(get_db)],
    order_repo: Annotated[OrderRepository, Depends(get_order_repo)],
    payment_gateway: Annotated[PaymentGateway, Depends(get_payment_gateway)],
    email_notifier: Annotated[EmailNotifier, Depends(get_email_notifier)],
    audit_logger: Annotated[AuditLogger, Depends(get_audit_logger)],
) -> dict[str, str]:
    return {"status": "canceled"}

2. Absence of Auto-Wiring & Cascading Factory Boilerplate

Beyond mitigating route signature pollution, real-world backend architectures fundamentally require composing dependencies within other dependencies (e.g., injecting a database session into a repository, which is then composed alongside gateways and notifiers into a domain use-case service).

We could try to clean up the route signatures by delegating execution to a single aggregated service:

@router.post("/orders/{order_id}/process")
async def process_order(
    order_id: int,
    order_uc_service: Annotated[OrderUcService, Depends(get_order_uc_service)],
) -> dict[str, str]:
    return await order_uc_service.process_order(order_id)

Unfortunately, this solves one problem only to create another.
Because FastAPI lacks type-driven auto-wiring, it cannot infer or build this dependency tree automatically from Python type annotations. Whenever you need to compose nested components, you are forced to write and maintain intermediate factory functions chained with explicit Depends() calls across every single layer of your architecture:

from typing import Annotated, AsyncGenerator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
 
# 1. Level-1 Factory: Database session lifecycle management
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with session_factory() as session:
        yield session
 
# 2. Level-2 Factory: Repository requires nested session resolution
def get_order_repo(
    db: Annotated[AsyncSession, Depends(get_db)]
) -> OrderRepository:
    return OrderRepository(session=db)
 
# (Additional Level-1 / Level-2 factories for PaymentGateway, 
# EmailNotifier, and AuditLogger must also be defined and maintained separately)
def get_payment_gateway() -> PaymentGateway: ...
def get_email_notifier() -> EmailNotifier: ...
def get_audit_logger() -> AuditLogger: ...
 
# 3. Top-Level Composite Factory: Manually wiring all 5 dependencies with Depends()
def get_order_uc_service(
    db: Annotated[AsyncSession, Depends(get_db)],
    order_repo: Annotated[OrderRepository, Depends(get_order_repo)],
    payment_gateway: Annotated[PaymentGateway, Depends(get_payment_gateway)],
    email_notifier: Annotated[EmailNotifier, Depends(get_email_notifier)],
    audit_logger: Annotated[AuditLogger, Depends(get_audit_logger)],
) -> OrderUcService:
    return OrderUcService(
        db=db,
        order_repo=order_repo,
        payment_gateway=payment_gateway,
        email_notifier=email_notifier,
        audit_logger=audit_logger,
    )

- The Architectural Bottleneck:
Signature pollution has not disappeared; it was merely displaced from the route handler to an intermediate factory function. Adding a single new infrastructure dependency (like a caching client or telemetry tracer) breaks this entire factory cascade and forces manual updates across every intermediate provider function.

3. Runtime Fail-Fast Absence & Silent Fallback Traps

FastAPI does not perform validation of the dependency graph during server startup. If a developer forgets Depends() on a parameter inside an intermediate factory, FastAPI fails silently at boot time and misinterprets the unannotated parameter as an incoming HTTP Request Body payload.

from typing import Annotated
from fastapi import APIRouter, Depends
 
router = APIRouter()
 
# ❌ The bug: 'db' lacks Depends(get_db).
# FastAPI silently treats 'db' as a required JSON body instead of raising an error at boot.
def get_order_service(db: DatabaseConnection) -> OrderService:
    return OrderService(db=db)
 
@router.get("/orders")
async def list_orders(
    service: Annotated[OrderService, Depends(get_order_service)]
) -> dict[str, str]:
    return {"status": "ok"}
# 1. At startup: No error is raised, server boots normally
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
 
# 2. At runtime: Standard GET requests fail unexpectedly with HTTP 422
HTTP/1.1 422 Unprocessable Entity
{
  "detail": [
    {
      "type": "missing",
      "loc": ["body", "db"],
      "msg": "Field required"
    }
  ]
}

4. Global Mutable State in Testing (dependency_overrides)

Overriding dependencies in test suites relies on mutating the global app.dependency_overrides dictionary. This design creates race conditions during parallel test execution (e.g., pytest -n auto) and leaks state across test cases if teardown fails.

async def test_order_creation(client: AsyncClient) -> None:
    # Global mutation: impacts all concurrent tests sharing the application instance
    app.dependency_overrides[get_payment_gateway] = lambda: MockPaymentGateway()
 
    response = await client.post("/orders", json={"amount": 100})
    assert response.status_code == 200
 
    # Manual teardown required to prevent state corruption in downstream tests
    app.dependency_overrides.clear()

5. Rigid Coarse-Grained Scopes & Missing Sub-Scopes

FastAPI only supports two lifecycle scopes: application-level singletons or per-request instances resolved via generators. It provides no native mechanism for sub-request scopes, transactional Unit of Work contexts, or long-lived user session scopes.

6. HTTP Protocol Lock-in & Non-Reusability

Because dependency resolution is coupled to Starlette's Request object lifecycle, dependency trees defined with Depends() cannot be reused in asynchronous background tasks (Taskiq/Celery), message consumers (Kafka/RabbitMQ), or CLI utilities.

Part 2: How Dishka Resolves Every Limitation

1. POJO Domain Services & Single-Point Injection

Dishka bridges FastAPI's functional routing with clean object-oriented domain design. All infrastructure dependencies are declared once in the constructor (__init__) of pure domain services. FastAPI routes only declare a single entry point (FromDishka[OrderService]). Adding a new dependency to OrderService requires zero changes to route handlers and zero intermediate factory functions.

from dishka.integrations.fastapi import FromDishka, inject
from fastapi import APIRouter
from sqlalchemy.ext.asyncio import AsyncSession
 
# 1. Pure domain service: 5 dependencies declared ONCE in __init__
class OrderService:
    def __init__(
        self,
        db: AsyncSession,
        order_repo: OrderRepository,
        payment_gateway: PaymentGateway,
        email_notifier: EmailNotifier,
        audit_logger: AuditLogger,
    ) -> None:
        self.db = db
        self.repo = order_repo
        self.payment = payment_gateway
        self.notifier = email_notifier
        self.logger = audit_logger
 
    async def process(self, order_id: int) -> None: ...
    async def cancel(self, order_id: int) -> None: ...
 
router = APIRouter()
 
# 2. Endpoints stay minimal: One entry point, zero infrastructure clutter
@router.post("/orders/{order_id}/process")
@inject
async def process_order(
    order_id: int,
    service: FromDishka[OrderService],
) -> dict[str, str]:
    await service.process(order_id)
    return {"status": "processed"}
 
@router.post("/orders/{order_id}/cancel")
@inject
async def cancel_order(
    order_id: int,
    service: FromDishka[OrderService],
) -> dict[str, str]:
    await service.cancel(order_id)
    return {"status": "canceled"}

2. Type-Based Auto-Wiring via Declarative Providers

Dishka inspects constructor type hints and automatically builds the dependency graph. Intermediate factory functions are eliminated; providers merely declare which classes and resources are available in the container.

from dishka import Provider, Scope, provide
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
class AppProvider(Provider):
    scope = Scope.REQUEST
 
    # Auto-wiring: Dishka inspects __init__ signatures and wires the graph automatically
    repo = provide(OrderRepository)
    service = provide(OrderService)
    payment = provide(StripePaymentGateway, provides=PaymentGateway)
    notifier = provide(EmailNotifier, scope=Scope.APP)
    logger = provide(AuditLogger, scope=Scope.APP)
 
    @provide
    async def provide_session(
        self, factory: async_sessionmaker[AsyncSession]
    ) -> AsyncIterable[AsyncSession]:
        async with factory() as session:
            yield session

3. Deterministic Fail-Fast DAG Validation at Startup

Dishka validates the complete Directed Acyclic Graph (DAG) during container initialization. Missing dependencies, cyclic references, and scope violations are caught before the server begins serving traffic.

from dishka import make_async_container
 
# Container bootstrap strictly validates all dependencies across all scopes
container = make_async_container(AppProvider())
dishka.exceptions.NoFactoryError: 
Cannot find factory for type <class 'PaymentGateway'> with scope <Scope.REQUEST>.
Requested by:
  <class 'OrderService'> (scope: Scope.REQUEST) via __init__(payment_gateway)

4. Hermetic Test Overrides Without Global State

Tests construct isolated container instances using override providers, eliminating global mutable state and guaranteeing thread-safe, concurrent test execution in CI/CD environments.

class MockPaymentProvider(Provider):
    @provide(scope=Scope.APP)
    def mock_payment(self) -> PaymentGateway:
        return AsyncMock(spec=PaymentGateway)
 
@pytest.fixture
async def test_container():
    # Composable, isolated container instance per test case
    container = make_async_container(AppProvider(), MockPaymentProvider())
    yield container
    await container.close()

5. Hierarchical Multi-Tier Scopes (APP, REQUEST, ACTION, STEP)

- Beyond Binary Lifecycles via Nested Containers:
Unlike frameworks restricted to global singletons or flat request scopes, Dishka allows hierarchical container nesting. A parent container can spawn child containers via async context managers. This enables fine-grained control over resource boundaries—such as isolating a short-lived transactional UnitOfWork inside a broader HTTP request scope without leaking state or keeping database sessions open longer than necessary.

- A Distinct Operational Role Across the Application Lifecycle:

  • Scope.APP: Lives for the entire application process (from startup to shutdown). Best suited for database connection pools, shared HTTP client sessions, and global immutable configuration.
  • Scope.REQUEST: Bound to a single incoming transport event (an HTTP request, a WebSocket connection, or an incoming message). Ideal for database sessions (AsyncSession), authenticated identity contexts, and request-bound telemetry.
  • Scope.ACTION: Scoped to an atomic unit of execution or business transaction within a request. Designed for isolating Unit of Work lifecycles, nested database transactions, or individual command executions that must commit/dispose independently.
  • Scope.STEP: The most granular tier, scoped to a single iteration or sub-phase within an algorithmic pipeline (e.g., handling an individual record during batch processing).

- Predefined Implicit Scope Ordering:
By default, Dishka organizes resource lifecycles into a strict, linear sequence of hierarchical tiers:

Scope.APP ──> Scope.REQUEST ──> Scope.ACTION ──> Scope.STEP

When entering an asynchronous context via async with parent_container() without specifying an explicit scope argument, Dishka automatically transitions one tier down the chain into the immediate child scope (e.g., opening a context from Scope.APP automatically instantiates Scope.REQUEST).

Below is an example to illustrate:

from dishka import make_async_container, Provider, Scope, provide
 
# 1. Root container (Scope.APP): lives for the entire server lifecycle
app_container = make_async_container(AppProvider())
 
async def handle_http_request():
    # 2. Entering Scope.REQUEST (spawned from APP container)
    async with app_container() as request_container:
 
        # 3. Spawning an inner sub-container for a narrow transaction boundary (Scope.ACTION)
        # Dishka resolves and manages an isolated UnitOfWork that auto-disposes upon exit.
        async with request_container() as action_container:
            uow = await action_container.get(UnitOfWork)
            await uow.commit()
 
        # The transaction is committed and closed here, 
        # while the HTTP request session remains active if needed.

6. Protocol-Agnostic Core Across Web, CLI, and Workers

Because the dependency container is decoupled from web framework internals, the exact same container definitions run without alteration across HTTP servers, asynchronous worker queues, and CLI commands.

from dishka import make_async_container
from dishka.integrations.fastapi import setup_dishka as setup_fastapi
from dishka.integrations.taskiq import setup_dishka as setup_taskiq
 
container = make_async_container(AppProvider())
 
# 1. Attached to FastAPI HTTP Server
app = FastAPI()
setup_fastapi(container, app)
 
# 2. Attached to Taskiq Background Worker
broker = InMemoryBroker()
setup_taskiq(container, broker)
Ce contenu a été publié dans Non classé. Vous pouvez le mettre en favoris avec ce permalien.

Laisser un commentaire

Votre adresse de messagerie ne sera pas publiée. Les champs obligatoires sont indiqués avec *