name

dependency-injection

description

FastAPI pattern for **app-scoped** dependency injection via `ResourceProvider[T]` classes. Each provider owns one resource's construction and teardown as a single `@asynccontextmanager`, declares dependencies on other providers via constructor arguments, and is exposed as a `fastapi.Depends` callable; providers are composed into the app with `compose_providers(*providers)`, which accepts `None` entries so conditional wiring inlines at the call site as `provider if cond else None`. "Resource" is load-bearing - this is *only* for things that live for the lifetime of the app. Per-request values (authenticated user, request-id, per-request DB transaction) use plain `async def` + `Depends`; `ResourceProvider` is not a universal DI container. Trigger whenever you add, refactor, or wire any app-scoped resource in a FastAPI project; HTTP clients, DB pools, service clients, Kafka producers/consumers, schedulers, caches, feature-flag clients, or any singleton that previously lived as a module-level global or inside `@app.on_event("startup")`. Also trigger when the user says "add a new service", "wire X into the app", "set up a lifespan resource", "replace this global", "this service depends on that one", "make this testable", "refactor startup code", or when a service needs another in its `__init__`.

Resource Management and Dependency Injection

Why the abstraction

Module-level singletons, @app.on_event("startup") handlers, and ad-hoc app.state assignments - the usual alternatives - share three problems:

1. Implicit wiring. Nothing enforces that A is built before B if B depends on A. Handler order, import-time side effects, and app.state keys are informal contracts.
2. No symmetric teardown. Shutdown handlers run independently — no LIFO order, no unwind on partial-startup failure.
3. Not injectable. A module-level redis_client = Redis(...) can't be swapped in tests without monkey-patching imports.

ResourceProvider[T] collapses construction, lifecycle, and dependency-exposure into one object, composed into the app lifespan in a single line.

The core primitive

A resource here means something has app-scoped lifecycle: built once at startup, shared across requests, torn down at shutdown. HTTP clients, connection pools, config, service objects.

One copy per project, typically at base/resource.py. Deliberately small — the only magic in the system.

"""Abstraction for app-scope resource management and dependency-injection"""

from abc import ABC, abstractmethod
from contextlib import (
    AbstractAsyncContextManager,
    AsyncExitStack,
    asynccontextmanager,
)
from typing import cast

from fastapi import FastAPI, Request


class ResourceProvider[T](ABC):
    def __init__(self, key: str | None = None):
        self._key = key or type(self).__name__

    @abstractmethod
    def provide(self, app: FastAPI) -> AbstractAsyncContextManager[T]:
        """Return an async context manager yielding the resource.
        Implement by decorating an ``async def`` with ``@asynccontextmanager``:
        construct before ``yield``, clean up after."""

    @asynccontextmanager
    async def lifespan(self, app: FastAPI):
        if hasattr(app.state, self._key):
            yield
            return
        async with self.provide(app) as value:
            setattr(app.state, self._key, value)
            try:
                yield
            finally:
                delattr(app.state, self._key)

    def inject(self, app: FastAPI) -> T:
        try:
            return cast(T, getattr(app.state, self._key))
        except AttributeError as exc:
            raise RuntimeError(
                f"{type(self).__name__} is not installed. "
                "Did you forget to include it in the app lifespan?"
            ) from exc

    def inject_optional(self, app: FastAPI) -> T | None:
        """Like ``inject``, but returns ``None`` when the provider wasn't
        installed — for downstreams that handle a conditionally-gated upstream."""
        return cast(T | None, getattr(app.state, self._key, None))

    def __call__(self, request: Request) -> T:
        return self.inject(request.app)


def compose_providers(*providers: ResourceProvider[object] | None):
    """Compose providers into a single app lifespan. ``None`` entries are
    skipped, so conditional wiring inlines at the call site as
    ``provider if cond else None`` — no runtime dispatch inside ``provide``."""
    @asynccontextmanager
    async def lifespan(app: FastAPI):
        async with AsyncExitStack() as stack:
            for p in providers:
                if p is not None:
                    await stack.enter_async_context(p.lifespan(app))
            yield

    return lifespan

Each part, briefly:

• provide(app) — the only abstract method. Decorate an async def with @asynccontextmanager: construct before yield, clean up after. Resolve upstream resources via other_provider.inject(app). If the resource is itself an async context manager (httpx.AsyncClient, a pool), delegate with async with ... as x: yield x.
• lifespan(app) — thin wrapper that stashes the yielded value on app.state and drops it on exit. Idempotent on _key collisions.
• inject(app) — typed accessor. Raises a specific RuntimeError if the provider wasn't installed, catching "forgot to wire it up" bugs at first use.
• inject_optional(app) — same lookup, but returns None instead of raising when the provider wasn't installed. The accessor choice is how a downstream declares "I require this upstream" vs "I handle its absence".
• __call__(request) — makes the provider itself a FastAPI dependency; Depends(provider) works because Depends accepts any callable with the right signature.
• _key — app.state attribute, defaults to class name. Override via __init__(key=...) only when you want two instances of the same provider class.
• compose_providers(*providers) — returns the lifespan= callable for FastAPI. None entries are skipped so feature-gated wiring stays inline at the call site. AsyncExitStack gives LIFO teardown and clean unwind on partial-startup failure.

Writing a provider

The service is lifecycle-independent — it doesn't read app.state, register hooks, or know the provider exists. Framework types like Request are fine as method arguments; they're just values.

# auth_service.py
class UserService:
    def __init__(self, api: ApiClient):
        self._api = api
        
    async def get_user(self, id):
      return await self._api.query(...)
    
        
class UserServiceProvider(ResourceProvider[UserService]):
    @asynccontextmanager
    async def provide(self, app: FastAPI):
        async with httpx.AsyncClient() as http:
            service = UserService(ApiClient(http))
            yield service

Rules for provide:

• All async initialization before yield; reaching yield means the resource is ready to serve.
• Cleanup after yield (in a finally: block, or via a nested async with that handles its own teardown).
• Don't swallow startup exceptions. A failing provide should abort app boot — that's what you want in production.
• No business logic, and no configuration branches — provide wires, the service computes. Feature flags and env toggles live at the entrypoint (see Configuration), never inside provide.

Providers with dependencies on other providers

Upstreams are declared as constructor arguments — other provider instances, not the built resources. At build time, resolve via inject:

class AuthClientProvider(ResourceProvider[AuthClient]):
    @asynccontextmanager
    async def provide(self, app: FastAPI):
        async with AuthClient(AuthConfig()) as client:
            yield client


class UserServiceProvider(ResourceProvider[UserService]):
    def __init__(self, auth_client_provider: AuthClientProvider):
        super().__init__()
        self._auth_client_provider = auth_client_provider

    @asynccontextmanager
    async def provide(self, app: FastAPI):
        auth = self._auth_client_provider.inject(app)
        yield UserService(auth_client=auth)

Reading the constructor signature tells you every dependency. No registry, no auto-wiring, no string keys.

Wiring and configuration

Three files, three roles:

• services/ — the provider class lives next to the service it wraps.
• layers/ — the provider singleton instance lives in a layer module, grouped with peers.
• app.py — composes layers into the lifespan and applies feature-flag gates.

This split is what lets routes Depends(provider) without ever importing app.py, and is how the import graph stays acyclic by construction.

The layer pattern

A layer is a module under layers/ that declares a cohesive group of provider singletons. A layer is not a class or runtime type — just a directory convention plus strict import rules.

services/               # ResourceProvider classes (UserServiceProvider, ...)
layers/                 # provider singletons, grouped by concern
  infra_layer.py        # HTTP clients, message buses, GCP clients
  service_layer.py      # auth_client_provider, user_service_provider
  ...
routers/                # import providers from layers/ for Depends(...)
app.py                  # imports layers/ for wiring only

Layer module anatomy — instantiate providers, export them individually, and collect them into a providers tuple so the compose call at app.py reads as a layer manifest:

# layers/infra_layer.py
from project.services.httpx_client import HttpxClientProvider

httpx_client_provider = HttpxClientProvider()

providers = (httpx_client_provider,)

Cross-layer dependencies are plain imports — a downstream layer depends on an upstream layer by importing the singleton it needs:

# layers/service_layer.py
from project.services.auth_client import AuthClientProvider
from project.services.user_service import UserServiceProvider

from .infra_layer import httpx_client_provider

auth_client_provider = AuthClientProvider(httpx_client_provider)
user_service_provider = UserServiceProvider(auth_client_provider)

providers = (auth_client_provider, user_service_provider)

Routes import from layers:

# routers/me.py
from project.layers.service_layer import user_service_provider

@router.get("/me")
async def me(user_service: Annotated[UserService, Depends(user_service_provider)]) -> User:
    return await user_service.get_me()

Import rules (what makes the graph acyclic):

• services/ imports only other services/, stdlib, and external libs.
• layers/ imports from services/ (to instantiate providers) and from other layers/ (upstream → downstream). Never from routers/ or app.py.
• routers/ imports from services/ (for types) and layers/ (for Depends). May import other routers for sub-router composition.
• app.py imports layers/ and routers/. Nothing imports app.py.

Layers form a DAG; app.py and routers/ are leaves. Circular imports are prevented by construction, not discipline.

Composing layers

app.py is the single wiring site. It imports layers, flattens their providers tuples into compose_providers, and applies the gates:

# app.py
from project.layers import infra_layer, service_layer

app = FastAPI(lifespan=compose_providers(
    *infra_layer.providers,
    *service_layer.providers,
))

Ordering matters. compose_providers builds in argument order and tears down in reverse. Within a layer, order providers upstream → downstream in the providers tuple. Across layers, list upstream layers first in the compose call. A missing upstream only blows up when the downstream's provide calls inject.

Configuration and feature flags

All configuration — flags, env toggles, region selectors — is resolved at app.py, to plain booleans, before compose_providers runs. provide() is never contextual: it doesn't read flags, inspect app, or branch on runtime state. That keeps each provider's lifecycle as predictable as a pure function.

Two rules keep optionality tidy:

1. Provider declarations and constructor dependencies are never None. Every UpstreamProvider parameter is a concrete, required provider — no | None, no defaults to None. Reading a constructor signature tells you which upstreams the provider expects to exist.
2. The compose call is the only place conditions live. compose_providers skips None entries, so a feature flag becomes a ternary at the argument position.

When a downstream's provide needs to handle an upstream that may not be installed, it uses inject_optional (returns T | None) rather than inject (raises). The accessor choice expresses at the call site whether the upstream is required:

class UserServiceProvider(ResourceProvider[UserService]):
    def __init__(
        self,
        auth_client_provider: AuthClientProvider,
        analytics_provider: AnalyticsProvider,
    ):
        super().__init__()
        self._auth_client_provider = auth_client_provider
        self._analytics_provider = analytics_provider

    @asynccontextmanager
    async def provide(self, app):
        auth = self._auth_client_provider.inject(app)              # required
        analytics = self._analytics_provider.inject_optional(app)  # optional
        yield UserService(auth, analytics)

Providers are declared in layers unconditionally; gates live only inside compose_providers:

# app.py
app = FastAPI(lifespan=compose_providers(
    *infra.providers,
    analytics.analytics_provider if settings.enable_analytics else None,
    infra.cache_provider         if not settings.debug_mode else None,
    *service.providers,
))

For groups — several providers enabled by the same flag — expand a layer conditionally:

app = FastAPI(lifespan=compose_providers(
    *infra.providers,
    *(analytics.providers if settings.enable_analytics else ()),
    *service.providers,
))

Reading the compose call, you can see the full graph and every flag that gates part of it. If a downstream's provide uses inject (not inject_optional) on an upstream that ends up gated out, you get the usual RuntimeError("... is not installed") at first request — exactly the loud failure you want for a mis-wired required dep.

Using a provider as a FastAPI dependency

Import the service type from services/ and the provider singleton from its layer:

from project.services.user_service import UserService
from project.layers.user import user_service_provider

async def get_current_user(
    # app-scoped, providers are pre-configured in the main entry
    user_service: Annotated[UserService, Depends(user_service_provider)],
) -> User:
    return await user_service.get_me()

@app.get("/me")
async def me(
    # request-scoped, via ordinary Depends
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    pass

FastAPI's async def + Depends already handles request-scoped values cleanly (authenticated user, request-id logger, per-request DB transaction) — this pattern does not replace that. The two scopes coexist in one handler:

Don't use the provider pattern for: per-request values (use async def + Depends), scripts / CLIs / workers that aren't FastAPI apps, or tiny apps with one or two resources where a plain lifespan function is enough.

What's the "service" btw?

Service and provider are always separate objects, in separate files. Same subpackage, fine; same class, no.

The service is the unit of behavior: plain Python, lifecycle-independent (no @app.on_event, no app.state reads, no self-registration), collaborators come in via the constructor, one cohesive responsibility. It is the unit of unit testing — if booting an app is required to test it, a lifecycle concern has leaked in.

Beneath it sits the model layer — pure domain code with zero external dependencies: validators, business rules, data classes, pure functions. Rules and calculations belong there, not in the service. The service orchestrates; the model computes. The model is the fastest tier to test (no mocks needed).

They don't necessarily named *Service - the suffix is just a rough naming when you cannot find a better name.

The provider is the unit of wiring: construct the service, own its lifespan, accept upstream providers as constructor arguments, contain no business logic. If provide does anything beyond assembling objects and starting their lifecycle, push the excess down to the service or model.

The two roles are deliberately opposite. Service: behavior-rich, lifecycle-independent, heavily tested. Provider: lifecycle-aware, behavior-free, barely tested.

Service shape: concrete class or Protocol + impl

One question decides: do you actually need more than one implementation of this service?

Case A (default): one concrete class, no Protocol

For services with exactly one production impl that are content to swap at the route boundary via app.dependency_overrides (see Testing). Least ceremony, type-checker clean, one class per service. Right for the majority of services in most apps.

class UserService:
    def __init__(self, auth_client: AuthClient):
        self._auth_client = auth_client

    async def get(self, user_id: str) -> User: ...


class UserServiceProvider(ResourceProvider[UserService]):
    def __init__(self, auth_client_provider: AuthClientProvider):
        super().__init__()
        self._auth_client_provider = auth_client_provider

    @asynccontextmanager
    async def provide(self, app):
        yield UserService(self._auth_client_provider.inject(app))

What you give up: ResourceProvider[UserService] is invariant on the concrete class, so you can't declare a FakeUserServiceProvider yielding a separate fake class without a type error. dependency_overrides doesn't hit this (typed loosely), so route-level tests still work.

Case B: body-less Protocol + impl

Upgrade when one of these is actually true:

• More than one production implementation (live vs replay, v1/v2, different backends, feature-flag A/B).
• A test needs the real downstream provider run against a fake upstream provider. dependency_overrides can't do this — it substitutes only at the outermost Depends of a route, not at inject lookups inside another provide.

If neither applies, stay in Case A. Promoting later is mechanical: extract methods into a Protocol, rename the concrete class to *Impl.

class UserService(Protocol):
    async def get(self, user_id: str) -> User: ...


class UserServiceImpl:                    # structurally satisfies UserService
    def __init__(self, auth_client: AuthClient):
        self._auth_client = auth_client

    async def get(self, user_id: str) -> User: ...


class FakeUserService:                    # same shape, no inheritance
    async def get(self, user_id): return User(id=user_id, ...)


class FakeUserServiceProvider(ResourceProvider[UserService]):
    @asynccontextmanager
    async def provide(self, app):
        yield FakeUserService()

Why body-less Protocol + separate impl: type checkers (mypy, pyright, ty) emit "Cannot instantiate protocol class" on any Protocol that has a body. The split keeps them quiet without giving up structural fakes.

Testing

Three levers, each at a different scope. Reach for the smallest one that covers the case.

1. Unit-test the service directly (always available) — construct with fakes, no app, no lifespan.

async def test_get_me():
    svc = UserService(FakeAuthClient(user_id="u1"))   # or UserServiceImpl in Case B
    assert await svc.get("u1") == User(id="u1", ...)

Model-layer code is simpler still: pure functions and data classes, no fakes at all.

2. Route test via app.dependency_overrides (Case A & B — the primary seam) — override by provider instance:

app.dependency_overrides[user_service_provider] = lambda: FakeUserService()
client = TestClient(app)
assert client.get("/me").status_code == 200
app.dependency_overrides.clear()   # or pytest fixture

dependency_overrides is typed Callable[..., Any], so the type checker doesn't enforce a shape match — duck typing carries you at runtime. Use when you're verifying a route's behavior (status codes, serialization, middleware) against a known service response, scoped to a single test.

3. Provider-seam test (Case B only) — build the real downstream against a fake upstream provider:

fake_auth_provider = FakeAuthClientProvider()
user_service_provider = UserServiceProvider(fake_auth_provider)   # the real one

test_app = FastAPI(
    lifespan=compose_providers(fake_auth_provider, user_service_provider),
)
test_app.include_router(user_router)

No app.state surgery. The downstream runs its real provide; only what it resolves via inject is the fake. Wanting this seam in Case A is the signal to promote to Case B.

Migration path

Greenfield wiring is the easy case. Real projects often carry legacy startup code — @app.on_event("startup") handlers, module-level globals, hand-managed app.state assignments — that can't be ripped out atomically. The goal of the migration path is to let DI-aware code depend on the ResourceProvider interface today while the legacy owner keeps managing lifecycle. Cut over later, piece by piece.

The bind_external pattern

Grow the provider with a setter that accepts an externally-owned instance. provide() branches: if something was bound, yield it; otherwise take the self-owning path.

class HttpxClientProvider(ResourceProvider[httpx.AsyncClient]):
    def __init__(self):
        super().__init__()
        self._external: Optional[httpx.AsyncClient] = None

    def bind_external(self, client: httpx.AsyncClient) -> None:
        """Migration-only: delegate to an externally-owned client.
        Delete once the legacy owner is gone."""
        self._external = client

    @asynccontextmanager
    async def provide(self, app: FastAPI):
        if self._external is not None:
            yield self._external
            return
        limits = httpx.Limits(max_connections=1000, max_keepalive_connections=100)
        async with httpx.AsyncClient(limits=limits, follow_redirects=True) as client:
            yield client

The singleton stays exported from its layer from day one:

# layers/infra.py
httpx_client_provider = HttpxClientProvider()   # supports bind_external during migration
providers = (httpx_client_provider,)

app.py orchestrates the bridge inside its lifespan: start the legacy owner, bind its resource into the provider, enter the composed DI lifespan:

# app.py
from project.layers import infra_layer
from project.legacy.httpx_client import httpx_client_wrapper

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.httpx_client_wrapper.start() # legacy owns lifecycle
    infra_layer.httpx_client_provider.bind_external(app.state.httpx_client_wrapper.async_client)
    try:
        async with compose_providers(*infra_layer.providers)(app):
            yield
    finally:
        await httpx_client_wrapper.stop()

app = FastAPI(lifespan=lifespan)

Downstream layers, route handlers, and tests only ever see ResourceProvider[httpx.AsyncClient]. They don't know whether the client is legacy-owned or self-owned — and therefore don't need to change when you cut over.

Cutover

When the last legacy caller is gone:

1. Delete bind_external and the self._external branch in provide().
2. Delete the legacy owner (httpx_client_wrapper, its start()/stop()).
3. lifespan collapses back to a plain compose_providers(...) call.

No downstream code changes. The import graph was stable throughout — that's the whole point of keeping singletons in layer modules rather than constructing them at the wiring site.

When the pattern does not apply

• Per-request legacy state (e.g., a middleware that stuffs something onto request.state): that's not an app-scoped resource, so ResourceProvider isn't the right tool. Migrate it to a plain async def dependency instead.

• Legacy that already exposes an async context manager: skip the setter and wrap directly inside provide():

  @asynccontextmanager
  async def provide(self, app):
      async with legacy_context_manager() as x:
          yield x

bind_external is the escape hatch for imperative legacy ownership (start()/stop() pairs), not a general-purpose extension point.

Common mistakes

• Defining provider singletons in app.py. Routes need to Depends(provider), app.py imports routes — you get a circular import. Singletons live in layers/ modules; app.py is a wiring site only.
• ResourceProvider for per-request values. Use async def + Depends instead.
• Forgetting @asynccontextmanager. Without it, provide is an async generator function; async with self.provide(app) will blow up.
• Work on the wrong side of yield. Before is setup and must finish before the app accepts traffic; after is teardown. Swapping them serves requests against a half-built resource or holds a connection for one request-lifetime.
• Branching on configuration inside provide. Config is resolved at the entrypoint; provide should be flag-free. Express optional wiring with provider if cond else None inside the compose_providers(...) call.
• Instantiating providers lazily inside provide. All providers must exist at module load and be listed in compose_providers(...); provider construction is cheap, I/O happens only inside provide.
• Reaching into app.state.XYZ directly. Use inject for its typed return and useful error. Same rule for request.state — route through a typed accessor.
• Forgetting a provider in compose_providers. Surfaces as RuntimeError("... is not installed") at first request; check the argument list.
• Circular provider dependencies. compose_providers builds strictly in argument order; break cycles by extracting shared state into a third provider.
• Work in __init__. Constructors only stash config / upstream providers. All async / I/O goes inside provide, between the decorator and yield.