Dependency injection¶

servicewright does not ship a DI container and does not depend on one. It only insists that there are two tiers:

A "unit of work" is one HTTP request, one RPC, one scheduled run, one consumed message. Same tier, same guarantees, regardless of which entrypoint produced it.

flowchart TD
    APP["AppScope — opened once at startup, closed last<br/>engine · pools · clients · caches"]
    APP --> U1["UnitScope<br/>HTTP request"]
    APP --> U2["UnitScope<br/>gRPC call"]
    APP --> U3["UnitScope<br/>job run"]
    APP --> U4["UnitScope<br/>consumed message"]
    U1 --> D1["session · use case"]
    U2 --> D2["session · use case"]
    U3 --> D3["session · use case"]
    U4 --> D4["session · use case"]

The contract¶

Three protocols, all structural — no inheritance, no registration:

class DependencyContainerProtocol(Protocol):
    def app_scope(self) -> AbstractAsyncContextManager[AppScopeProtocol]: ...
    def unit_scope(self, context: Mapping | None = None) -> AbstractAsyncContextManager[UnitScopeProtocol]: ...


class AppScopeProtocol(Protocol):
    async def get(self, dependency_key: type[T] | str) -> T: ...


class UnitScopeProtocol(Protocol):
    async def get(self, dependency_key: type[T] | str) -> T: ...

Two context managers and a get. That is the entire integration surface, which is why a hand-written container fits in twenty lines:

class Container:
    def __init__(self, provides: Mapping[Any, Any]) -> None:
        self._provides = dict(provides)

    @contextlib.asynccontextmanager
    async def app_scope(self) -> AsyncIterator[Scope]:
        yield Scope(self._provides)

    @contextlib.asynccontextmanager
    async def unit_scope(self, context: Mapping | None = None) -> AsyncIterator[Scope]:
        yield Scope(self._provides)

In practice you use dishka, where Scope.APP and Scope.REQUEST map straight onto the two tiers.

Who opens the unit scope¶

You almost never call unit_scope() yourself. Each entrypoint opens one at the right moment and hands it to you:

The context mapping is handed to your container, which decides what to do with it — a hand-written container might stash it, log it, or use it to pick a shard.

Reaching the scope from your code¶

from servicewright.adapters.fastapi import UnitScopeDep

@router.get("/orders/{order_id}")
async def get_order(order_id: str, scope: UnitScopeDep) -> dict:
    use_case = await scope.get(GetOrder)
    return await use_case.execute(order_id)

from litestar import get
from litestar.di import Provide

from servicewright.adapters.litestar import get_unit_scope

@get("/orders/{order_id:str}", dependencies={"unit_scope": Provide(get_unit_scope)})
async def get_order(order_id: str, unit_scope: UnitScopeProtocol) -> dict:
    use_case = await unit_scope.get(GetOrder)
    return await use_case.execute(order_id)

from servicewright.adapters.grpc import current_unit_scope

class OrdersServicer(orders_pb2_grpc.OrdersServicer):
    async def GetOrder(self, request, context):
        scope = current_unit_scope()
        use_case = await scope.get(GetOrder)
        return await use_case.execute(request.order_id)

async def sweep_expired_orders(scope: UnitScopeProtocol) -> None:
    orders = await scope.get(OrderRepository)
    await orders.delete_expired()

The application scope¶

Opened once, closed last. Entrypoints receive it on their ServiceContext:

async def bind(self, ctx: ServiceContext) -> None:
    pool = await ctx.app_scope.get(ConnectionPool)

Lifecycle hooks receive it too. Closing it is the final step of shutdown, after every entrypoint has drained and stopped, which is what guarantees an in-flight request never finds its pool already closed.

Writing your own adapter¶

Implement the two context managers and you are done. Two rules:

1. app_scope() must finalize on exit. Whatever your container does to close pools and run finalizers has to happen when the context manager exits, because the Host relies on that as the last cleanup step.
2. unit_scope() must be independent per call. Entrypoints open many of them concurrently.

Next¶

• dishka adapter — the batteries-included option.
• Entrypoints — ServerEntrypoint vs ScopedEntrypoint, and why the split exists.