FastAPI Dependency Injection: Scoped Resources, Testing, Overrides

What is FastAPI dependency injection, and how does Depends resolve the graph?

FastAPI's dependency injection is not a container you register services with. There is no global registry, no lifetimes declared up front, and no interface-to-implementation binding. Instead, when a route is registered, FastAPI inspects the signature of the path operation function, finds every parameter declared with Depends(...), then recursively inspects those callables' signatures. The result is a tree of callables attached to the route. On each request, FastAPI walks that tree and calls each node, so a dependency's inputs are always resolved before it runs.

Consider a chain of three: get_settings returns configuration, get_db takes settings and opens a session, and get_current_user takes the session plus a bearer token and loads the user. The endpoint only asks for get_current_user. FastAPI resolves settings first, then the session, then the user, and finally calls the endpoint. You declare what each function needs, and the order follows from that.

Within a single request, results are cached by dependency callable. If the endpoint also declares get_db directly, it receives the same session object that get_current_user used, not a second one. This is usually what you want, because both share one transaction and one connection.

The catch is that caching is silent. If a dependency is supposed to run twice, for example to produce two independent sessions or two distinct generated IDs, it runs once and both parameters get the same value. Pass Depends(dep, use_cache=False) at the call site that needs a fresh invocation.

Copyfrom typing import Annotated
from uuid import uuid4
from fastapi import Depends, FastAPI, Header

app = FastAPI()

class Settings:
    database_url = "postgresql://app:changeme@localhost:5432/appdb"

class Session:  # stand-in for a real DB session
    def __init__(self, url: str): self.url = url
    def close(self): pass

def get_settings() -> Settings:
    return Settings()

def get_db(settings: Annotated[Settings, Depends(get_settings)]):
    session = Session(settings.database_url)  # runs after get_settings
    try:
        yield session
    finally:
        session.close()

def get_current_user(db: Annotated[Session, Depends(get_db)],
                     authorization: Annotated[str, Header()]) -> dict:
    # A real app would validate the bearer token and query the user via db
    return {"name": "alice", "session_id": id(db)}

def new_id() -> str:
    return uuid4().hex

@app.get("/me")
def read_me(
    user: Annotated[dict, Depends(get_current_user)],
    db: Annotated[Session, Depends(get_db)],  # cached: same session the user dependency got
    id_a: Annotated[str, Depends(new_id)],
    id_b: Annotated[str, Depends(new_id)],  # cached: silently equals id_a
    id_c: Annotated[str, Depends(new_id, use_cache=False)],  # forces a fresh call
):
    return {"same_session": user["session_id"] == id(db),
            "a_equals_b": id_a == id_b, "a_equals_c": id_a == id_c}

A settings -> session -> user chain where per-request caching makes the endpoint share one session and one generated ID, and use_cache=False forces a separate call.

How do yield dependencies manage request-scoped resources?

Code before the yield is setup: open a database session, check out a connection, start a transaction. The yielded value is what the endpoint receives. Code after the yield is teardown, and FastAPI runs it once the request is finished with the resource. Setup and teardown sit in one function, so the code that acquires a resource also releases it, and the endpoint never has to.

Wrap the yield in try/except/finally. If the endpoint raises, FastAPI throws that exception into the generator at the yield point, the same way contextlib.contextmanager does. Your except block can then roll back the transaction instead of committing half-written state. The finally block closes the session or returns the connection to the pool, whether the request succeeded or failed. If you omit finally, an unexpected exception type can skip cleanup and leak connections until the pool runs out.

The except block must end with a bare raise. If you catch the exception and let the generator finish normally, you hide the original error from FastAPI's exception handlers and from your logs. The client then gets a generic server error or a response that doesn't match what happened, and you have no traceback showing the cause. Roll back, then re-raise.

When teardown runs has changed across FastAPI releases. In some versions it runs before the response is sent, and in others after. Its order relative to background tasks has changed too. This determines whether a background task can still use the session and whether teardown can still change the response. Check the release notes for the FastAPI version you deploy, and don't assume a particular order.

Copyfrom typing import Annotated

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

# Placeholder DSN; pool_size caps how many connections can leak before exhaustion.
engine = create_engine("postgresql://app:changeme@db.internal:5432/orders", pool_size=5)
SessionLocal = sessionmaker(bind=engine)

app = FastAPI()


def get_session():
    # Setup: check out a session before the endpoint runs.
    session = SessionLocal()
    try:
        yield session  # The endpoint receives this value.
        # Reached only if the endpoint returned without raising.
        session.commit()
    except Exception:
        # The endpoint's exception is thrown in at the yield point.
        session.rollback()  # Discard half-written state.
        raise  # Re-raise so exception handlers and logs see the real error.
    finally:
        # Runs on success and failure: return the connection to the pool.
        session.close()


SessionDep = Annotated[Session, Depends(get_session)]


@app.post("/orders/{order_id}/cancel")
def cancel_order(order_id: int, session: SessionDep):
    order = session.get(Order, order_id)  # Order: your mapped model
    if order is None:
        raise ValueError("order not found")  # Triggers rollback, then propagates.
    order.status = "cancelled"
    return {"id": order_id, "status": order.status}

A yield dependency that opens a SQLAlchemy session as setup, commits on success, rolls back and re-raises on failure, and always closes the session in finally.

Where should app-scoped and request-scoped resources live?

Resources fall into two lifetimes, and mixing them up is one of the most common causes of production incidents in FastAPI services. Anything expensive to construct and safe to share, such as a database connection pool, an HTTP client, or parsed configuration, belongs to the application. Create it once in the lifespan handler, store it on app.state, and close it when the lifespan exits. Anything that carries per-request state, such as a database session or an open transaction, belongs to the request. Expose it through a yield dependency that reads the shared pool from request.app.state, checks out a connection, and returns it during teardown.

The failure mode appears when a pool or client is built inside a yield dependency. Every request then pays for TCP and TLS handshakes and database authentication. It also discards keep-alive connections, so nothing is reused. Under load, the request rate turns directly into connection opens. That can push the database past its connection limit and fill the host with sockets in TIME_WAIT. It also exposes any client that is not closed cleanly, which leaks file descriptors. Latency degrades first, and then requests begin to fail.

Use the following rule to choose a scope. If two concurrent requests can safely share the object, it is application-scoped. If the object holds state that one request could corrupt for another, such as uncommitted writes or a user identity, it is request-scoped.

Copyfrom contextlib import asynccontextmanager
from typing import AsyncIterator

import asyncpg
import httpx
from fastapi import Depends, FastAPI, Request


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # App-scoped: build these once at startup so every request reuses their connections.
    app.state.db_pool = await asyncpg.create_pool(
        dsn="postgresql://app:change-me@db.internal:5432/orders", min_size=2, max_size=10
    )
    app.state.http = httpx.AsyncClient(base_url="", timeout=5.0)
    try:
        yield
    finally:
        # Close shared resources once, at shutdown.
        await app.state.http.aclose()
        await app.state.db_pool.close()


app = FastAPI(lifespan=lifespan)


async def get_conn(request: Request) -> AsyncIterator[asyncpg.Connection]:
    # Request-scoped: borrow a pooled connection and wrap the request in a transaction.
    # The connection goes back to the pool on teardown. It is not closed.
    async with request.app.state.db_pool.acquire() as conn:
        async with conn.transaction():
            yield conn


def get_http(request: Request) -> httpx.AsyncClient:
    # Hand out the shared client. Never construct a new AsyncClient per request.
    return request.app.state.http


@app.get("/orders/{order_id}")
async def read_order(order_id: int, conn=Depends(get_conn), http=Depends(get_http)):
    row = await conn.fetchrow("SELECT id, status FROM orders WHERE id = $1", order_id)
    resp = await http.get(f"/tracking/{order_id}")
    return {"id": row["id"], "status": row["status"], "tracking": resp.json()}

The asyncpg pool and httpx client are created once in the lifespan handler and stored on app.state, while a yield dependency borrows a pooled connection for each request, opens a transaction, and returns the connection to the pool during teardown.

Should dependencies be sync or async?

The scope decision tells you how long a resource lives. The def versus async def decision tells you where its code runs, and choosing wrong produces different failures.

FastAPI awaits an async def dependency directly on the event loop. It calls a plain def dependency through a worker threadpool so it cannot block the loop. That pool is finite and shared with sync path operations, so each sync dependency holds a thread slot until it returns.

An async def dependency that does no I/O is cheaper than a def one because it skips the thread hop. Validating a header or reading a value from request.state both fit this case.

Thread-bound resources need extra care. FastAPI does not guarantee that a sync dependency, the endpoint that uses it, and a sync yield dependency's teardown all run on the same thread. A resource that is tied to the thread that created it can fail at teardown or behave unpredictably. Examples include a SQLite connection with same-thread checking, or anything that stores state in thread-locals. Either create and close such resources in one call, or configure them to tolerate cross-thread use.

How do you build reusable, parameterized dependencies with Annotated?

Depends accepts any callable, so a dependency does not have to be a plain function. A class instance that defines __call__ works too: FastAPI inspects the __call__ signature, not __init__, so constructor arguments become configuration while __call__ parameters are resolved per request. A RequireRole("admin") instance can hold the role name and check it against the current user on every call. A factory function does the same job with less ceremony: require_role("admin") returns an inner function that closes over the role. Both turn a check you would otherwise copy into each handler into one tested unit with a parameter.

Signatures get noisy when every endpoint spells out user: User = Depends(get_current_user). A type alias such as CurrentUser = Annotated[User, Depends(get_current_user)] moves the wiring into one place. Handlers then declare user: CurrentUser. If you later swap the underlying dependency, you change one line instead of every route. Type checkers and editors still see User.

Some concerns produce no value the handler needs, such as authentication gates, rate limiting and audit logging. Attach these through dependencies=[...] on an APIRouter, on include_router, or on the FastAPI app itself. They run for every route in scope and their return values are discarded. A new route added to an admin router inherits the admin check automatically, so nobody can forget it.

These patterns also make review cheaper. A reviewer checks the policy once, in one module. After that, confirming a route is protected only means confirming which router it lives on. As the route count grows, you avoid repeating that review for every endpoint.

Copyfrom typing import Annotated
from fastapi import APIRouter, Depends, FastAPI, Header, HTTPException
from pydantic import BaseModel

class User(BaseModel):
    name: str
    roles: set[str]

def get_current_user(x_user: Annotated[str | None, Header()] = None) -> User:
    # Placeholder lookup; real code would validate a token here.
    if x_user is None:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return User(name=x_user, roles={"admin"} if x_user == "alice" else set())

# One alias holds the wiring. Swap the dependency here, not in every route.
CurrentUser = Annotated[User, Depends(get_current_user)]

class RequireRole:
    # __init__ takes configuration. FastAPI resolves __call__'s params per request.
    def __init__(self, role: str):
        self.role = role
    def __call__(self, user: CurrentUser) -> User:
        if self.role not in user.roles:
            raise HTTPException(status_code=403, detail=f"Requires {self.role}")
        return user

def require_role(role: str):
    # Factory alternative: the inner function closes over `role`.
    return RequireRole(role).__call__

def rate_limit() -> None:
    pass  # Placeholder: check a per-client counter and raise 429 when over the limit.

# Every route on this router inherits the admin gate. Return values are discarded.
admin = APIRouter(prefix="/admin", dependencies=[Depends(require_role("admin"))])

@admin.get("/stats")
def stats(user: CurrentUser):
    return {"requested_by": user.name}

app = FastAPI(dependencies=[Depends(rate_limit)])  # App-wide: applies to all routes
app.include_router(admin)

A class-based RequireRole dependency, a factory wrapper and a CurrentUser Annotated alias, with an admin check attached at router level and a rate limit attached at app level.

How do you test FastAPI apps with dependency overrides?

FastAPI exposes app.dependency_overrides, a plain dict that maps the original dependency callable to a replacement. Resolution uses the function object itself as the key, so you must import and override the exact callable the routes reference. A lambda or wrapper that merely looks equivalent will not match.

Override at the narrowest seam. Replacing get_session or get_payment_client keeps validation, serialization, and endpoint logic under test. Overriding a high-level service dependency skips the code you most want to exercise.

Override auth the same way. If a router declares dependencies=[Depends(require_user)], overriding require_user swaps it on every route of that router, because router- and app-level dependencies are resolved through the same mechanism as parameter dependencies. Have the replacement return a fixed test principal so you can still assert on authorization branches.

Clear overrides in fixture teardown, after the yield, so a failing test cannot leak its fakes into the next one. For databases, open a connection, begin a transaction, and bind the overridden session to that connection. Roll back in teardown. Each test sees a clean state without truncating tables.

Startup and shutdown logic in the lifespan handler runs only when TestClient is used as a context manager (with TestClient(app) as client:). Constructed without with, app-scoped resources are never initialized.

Overrides are global mutable state on the app object. Tests that share one module-level app and run concurrently in the same process will see each other's overrides, which produces flaky, order-dependent failures. Build a fresh app per test from a factory function instead.

When should you avoid relying on Depends, and what are the alternatives?

Depends covers request handling well, but it has limits. It has no singleton scope and no custom scopes. Every dependency is resolved per request, and results are cached only within that request. It also exists only inside FastAPI's request handling. A queue consumer, a cron job or a CLI migration script cannot ask the framework to resolve a dependency graph. Those entry points must call the underlying functions directly or share a separate wiring layer with the HTTP app.

Deep chains cause problems of their own. When a route parameter pulls in five nested dependencies, the signature no longer shows the database round-trips, the remote calls or the order they run in. Profiling then becomes the only way to understand what one request costs.

Tests carry a related risk. If every collaborator is overridden, the suite can pass while the real graph is broken: a wrong settings key, a missing teardown, or a session bound to the wrong engine. Keep a small set of integration tests that run the unmodified graph against real or containerized services.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to fastapi dependency injection, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Changing a shared variable is one edit and one review instead of editing every service's .env by hand.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• A wrong or stale value is caught at review time by the services it affects, not by a failed deploy.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.