Updated July 1, 2026

On this page

Guide to fastware DI: per-request resolution with caching, sync/async factories, generator cleanup, router deps, and test overrides.

#Dependency Injection

fastware's DependencyResolver provides per-request dependency resolution with caching and automatic cleanup. It supports sync and async factory functions, the yield pattern for resource lifecycle management, and dependency overrides for testing.

#What DI solves

Web handlers often need shared resources -- database connections, authenticated user objects, configuration, external API clients. Managing these manually leads to 3 common problems that grow worse as the application scales. Dependency injection solves all 3 by letting the framework create, cache, and clean up resources per request. Without DI, you either:

Create resources inside each handler (wasteful, no sharing)
Use module-level globals (hard to test, no per-request lifecycle)
Pass everything through middleware and request state (boilerplate)

DI lets you declare what a handler needs, and the framework resolves it once per request, caches it (so the same factory called twice returns the same instance), and cleans it up when the request is done. Generator factories are cleaned up in reverse order -- last resolved, first cleaned.

#Basic usage

#1. Define factory functions

A factory function receives a Request and returns the dependency value. Factories can be either sync or async, and can optionally use the yield pattern for resource lifecycle management with automatic cleanup:

python

async def get_db(request):
    return await create_connection(DATABASE_URL)

Factories that don't need request context can omit the parameter:

python

async def get_config():
    return load_config()

#2. Declare dependencies on routes

Pass a deps dict mapping parameter names to factory callables on any route decorator. The resolver calls each factory once per request, caches the result, and injects it as a keyword argument:

python

from fastware import Router

router = Router()

@router.get("/users", deps={"db": get_db})
async def list_users(request, db):
    rows = await db.fetch_all("SELECT * FROM users")
    return [dict(r) for r in rows]

The resolved dependency is passed as a keyword argument matching the dict key. The handler signature must accept the parameter name.

#3. Create the app

python

from fastware import create_app

app = create_app(router)

The app factory creates a DependencyResolver internally and uses it to resolve route dependencies on each request.

python

class TestDIIntegrationHTTP:
    """DI resolution wired into HTTP handler dispatch."""

    @pytest.mark.anyio
    async def test_sync_dep_passed_to_handler(self, client_for):
        router = Router()

        def get_user(request):
            return {"name": "alice"}

        @router.get("/profile", deps={"user": get_user})
        async def profile(request, user=None):
            return JSONResponse({"user": user})

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/profile")
        assert resp.status_code == 200
        assert resp.json() == {"user": {"name": "alice"}}

    @pytest.mark.anyio
    async def test_async_dep_passed_to_handler(self, client_for):
        router = Router()

        async def get_user(request):
            return {"name": "bob"}

        @router.get("/profile", deps={"user": get_user})
        async def profile(request, user=None):
            return JSONResponse({"user": user})

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/profile")
        assert resp.status_code == 200
        assert resp.json() == {"user": {"name": "bob"}}

    @pytest.mark.anyio
    async def test_generator_dep_with_cleanup(self, client_for):
        """Generator dep cleanup runs after handler returns."""
        cleanup_ran = False

        def get_conn(request):
            nonlocal cleanup_ran
            try:
                yield "test_connection"
            finally:
                cleanup_ran = True

        router = Router()

        @router.get("/data", deps={"conn": get_conn})
        async def data(request, conn=None):
            return JSONResponse({"conn": conn})

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/data")
        assert resp.status_code == 200
        assert resp.json() == {"conn": "test_connection"}
        assert cleanup_ran

    @pytest.mark.anyio
    async def test_async_generator_dep_with_cleanup(self, client_for):
        cleanup_ran = False

        async def get_conn(request):
            nonlocal cleanup_ran
            try:
                yield "async_test_conn"
            finally:
                cleanup_ran = True

        router = Router()

        @router.get("/data", deps={"conn": get_conn})
        async def data(request, conn=None):
            return JSONResponse({"conn": conn})

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/data")
        assert resp.status_code == 200
        assert resp.json() == {"conn": "async_test_conn"}
        assert cleanup_ran

    @pytest.mark.anyio
    async def test_two_deps_one_sync_one_async_gen(self, client_for):
        """Handler receives both sync and async generator deps."""
        gen_cleanup = False

        def get_config(request):
            return {"debug": True}

        async def get_conn(request):
            nonlocal gen_cleanup
            try:
                yield "db_conn"
            finally:
                gen_cleanup = True

        router = Router()

        @router.get("/both", deps={"config": get_config, "conn": get_conn})
        async def both(request, config=None, conn=None):
            return JSONResponse({"config": config, "conn": conn})

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/both")
        assert resp.status_code == 200
        data = resp.json()
        assert data["config"] == {"debug": True}
        assert data["conn"] == "db_conn"
        assert gen_cleanup

    @pytest.mark.anyio
    async def test_no_deps_handler_still_works(self, client_for):
        """Handlers without deps are unaffected by DI wiring."""
        router = Router()

        @router.get("/health")
        async def health(request):
            return JSONResponse({"ok": True})

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/health")
        assert resp.status_code == 200
        assert resp.json() == {"ok": True}

    @pytest.mark.anyio
    async def test_router_level_deps_merged_with_handler_deps(self, client_for):
        """Router-level deps from include_router merge with per-handler deps."""
        def get_user(request):
            return "admin"

        def get_config(request):
            return {"level": "high"}

        sub = Router()

        @sub.get("/info", deps={"config": get_config})
        async def info(request, user=None, config=None):
            return JSONResponse({"user": user, "config": config})

        main = Router()
        main.include_router(sub, prefix="/api", deps={"user": get_user})

        app = create_app(main)
        async with client_for(app) as client:
            resp = await client.get("/api/info")
        assert resp.status_code == 200
        data = resp.json()
        assert data["user"] == "admin"
        assert data["config"] == {"level": "high"}

    @pytest.mark.anyio
    async def test_handler_deps_override_router_deps(self, client_for):
        """Per-handler deps take priority over router-level deps with the same name."""
        def router_user(request):
            return "router_level"

        def handler_user(request):
            return "handler_level"

        sub = Router()

        @sub.get("/who", deps={"user": handler_user})
        async def who(request, user=None):
            return JSONResponse({"user": user})

        main = Router()
        main.include_router(sub, prefix="/api", deps={"user": router_user})

        app = create_app(main)
        async with client_for(app) as client:
            resp = await client.get("/api/who")
        assert resp.status_code == 200
        assert resp.json()["user"] == "handler_level"

    @pytest.mark.anyio
    async def test_cleanup_runs_on_handler_exception(self, client_for):
        """Generator cleanup runs even when the handler raises."""
        cleanup_ran = False

        def get_conn(request):
            nonlocal cleanup_ran
            try:
                yield "conn"
            finally:
                cleanup_ran = True

        router = Router()

        @router.get("/fail", deps={"conn": get_conn})
        async def fail(request, conn=None):
            raise RuntimeError("handler error")

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/fail")
        assert resp.status_code == 500
        assert cleanup_ran

#Sync and async factories

Both sync and async factory functions are supported. The resolver detects whether a factory returns an awaitable and handles it automatically, so you can mix sync and async factories freely in the same route's deps dict:

python

# Sync factory
def get_settings(request):
    return Settings(debug=request.header("X-Debug") == "true")

# Async factory
async def get_db(request):
    return await create_connection(DATABASE_URL)

If a factory returns an awaitable, the resolver awaits it automatically.

python

class TestDependencyResolverSync:
    """Sync dependency factory resolved and passed to handler."""

    @pytest.mark.anyio
    async def test_sync_factory(self):
        def get_db(request):
            return "db_connection"

        resolver = DependencyResolver()
        resolved, cleanups = await resolver.resolve(
            {"db": get_db}, "fake_request",
        )
        assert resolved == {"db": "db_connection"}
        assert cleanups == []

    @pytest.mark.anyio
    async def test_multiple_sync_factories(self):
        def get_db(request):
            return "db_conn"

        def get_cache(request):
            return "redis_conn"

        resolver = DependencyResolver()
        resolved, cleanups = await resolver.resolve(
            {"db": get_db, "cache": get_cache}, "fake_request",
        )
        assert resolved == {"db": "db_conn", "cache": "redis_conn"}
        assert cleanups == []

python

class TestDependencyResolverAsync:
    """Async dependency factory resolved."""

    @pytest.mark.anyio
    async def test_async_factory(self):
        async def get_db(request):
            return "async_db"

        resolver = DependencyResolver()
        resolved, cleanups = await resolver.resolve(
            {"db": get_db}, "fake_request",
        )
        assert resolved == {"db": "async_db"}
        assert cleanups == []

#Generator factories (yield pattern)

For resources that need cleanup (database connections, file handles, transactions), use the yield pattern. The factory yields the value, and the code after yield runs automatically after the handler completes, even if the handler raises an exception. This ensures resources are never leaked:

#Sync generator

python

def get_db_session(request):
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

#Async generator

python

async def get_db(request):
    conn = await asyncpg.connect(DATABASE_URL)
    try:
        yield conn
    finally:
        await conn.close()

Cleanup runs in reverse order (last resolved, first cleaned up). Cleanup errors are suppressed -- a failing cleanup does not mask the handler's response or exception.

python

class TestDependencyResolverGenerators:
    """Generator dependency with cleanup (yield pattern)."""

    @pytest.mark.anyio
    async def test_sync_generator_cleanup(self):
        cleanup_ran = False

        def get_conn(request):
            nonlocal cleanup_ran
            conn = "sync_conn"
            try:
                yield conn
            finally:
                cleanup_ran = True

        resolver = DependencyResolver()
        resolved, cleanups = await resolver.resolve(
            {"conn": get_conn}, "fake_request",
        )
        assert resolved == {"conn": "sync_conn"}
        assert len(cleanups) == 1
        assert not cleanup_ran

        await resolver.cleanup(cleanups)
        assert cleanup_ran

    @pytest.mark.anyio
    async def test_async_generator_cleanup(self):
        cleanup_ran = False

        async def get_conn(request):
            nonlocal cleanup_ran
            conn = "async_conn"
            try:
                yield conn
            finally:
                cleanup_ran = True

        resolver = DependencyResolver()
        resolved, cleanups = await resolver.resolve(
            {"conn": get_conn}, "fake_request",
        )
        assert resolved == {"conn": "async_conn"}
        assert len(cleanups) == 1
        assert not cleanup_ran

        await resolver.cleanup(cleanups)
        assert cleanup_ran

    @pytest.mark.anyio
    async def test_cleanup_runs_in_reverse_order(self):
        order = []

        def dep_a(request):
            try:
                yield "a"
            finally:
                order.append("a_cleanup")

        def dep_b(request):
            try:
                yield "b"
            finally:
                order.append("b_cleanup")

        resolver = DependencyResolver()
        resolved, cleanups = await resolver.resolve(
            {"a": dep_a, "b": dep_b}, "fake_request",
        )
        assert resolved == {"a": "a", "b": "b"}
        await resolver.cleanup(cleanups)
        # Reverse order: b cleaned up first, then a
        assert order == ["b_cleanup", "a_cleanup"]

    @pytest.mark.anyio
    async def test_cleanup_error_suppressed(self):
        """Cleanup errors are swallowed -- they must not mask the response."""

        def bad_cleanup(request):
            try:
                yield "value"
            finally:
                raise RuntimeError("cleanup failed")

        resolver = DependencyResolver()
        resolved, cleanups = await resolver.resolve(
            {"val": bad_cleanup}, "fake_request",
        )
        assert resolved == {"val": "value"}
        # Should not raise
        await resolver.cleanup(cleanups)

#Per-request caching

When the same factory is used in multiple deps on the same request, it is resolved exactly once and the cached result is reused for all subsequent references. This prevents creating duplicate database connections or re-authenticating on every dependency resolution:

python

async def get_db(request):
    return await create_connection(DATABASE_URL)

async def get_user_repo(request):
    # If get_db is also in this request's deps, it returns the same connection
    return UserRepository(await create_connection(DATABASE_URL))

Caching is based on the factory function's identity (id(factory)). If two dependency names point to the same factory callable, they get the same resolved value.

python

class TestDependencyResolverCaching:
    """Two handlers sharing the same dep get the same cached instance."""

    @pytest.mark.anyio
    async def test_same_factory_cached(self):
        call_count = 0

        def get_conn(request):
            nonlocal call_count
            call_count += 1
            return f"conn_{call_count}"

        resolver = DependencyResolver()
        resolved, _ = await resolver.resolve(
            {"conn1": get_conn, "conn2": get_conn}, "fake_request",
        )
        # Same factory -> same instance, called only once
        assert call_count == 1
        assert resolved["conn1"] == resolved["conn2"]
        assert resolved["conn1"] == "conn_1"

    @pytest.mark.anyio
    async def test_different_factories_not_cached(self):
        def get_a(request):
            return "a"

        def get_b(request):
            return "b"

        resolver = DependencyResolver()
        resolved, _ = await resolver.resolve(
            {"a": get_a, "b": get_b}, "fake_request",
        )
        assert resolved == {"a": "a", "b": "b"}

#Router-level dependencies

Apply dependencies to all routes in a sub-router using include_router, so every route under that prefix automatically receives the specified dependencies without repeating the deps dict on each route decorator:

python

api_router = Router()

@api_router.get("/items")
async def list_items(request, db):
    return await db.fetch_all("SELECT * FROM items")

@api_router.get("/items/{id:int}")
async def get_item(request, db):
    item_id = request.path_params["id"]
    return await db.fetch_one("SELECT * FROM items WHERE id = $1", item_id)

main_router = Router()
main_router.include_router(api_router, prefix="/api", deps={"db": get_db})

Router-level deps are merged with per-route deps. Per-route deps override router-level deps if there's a name conflict.

Handlers that don't accept a particular dependency keyword are not affected -- the resolver filters resolved deps to only those the handler's signature accepts.

#Dependency overrides for testing

Replace factory functions with test doubles using dependency_overrides on create_app. This lets you substitute real databases, API clients, and auth providers with in-memory fakes during testing, without changing any route code:

python

from fastware import create_app, AppConfig

# Production factory
async def get_db(request):
    return await create_connection(REAL_DATABASE_URL)

# Test factory
async def get_test_db():
    return FakeDatabase()

# Create app with overrides
app = create_app(router, config=AppConfig(
    dependency_overrides={get_db: get_test_db},
))

When the resolver encounters get_db in a route's deps, it calls get_test_db instead. The override factory can omit the request parameter if it doesn't need it.

This integrates with fastware's test client:

python

from fastware.testing import TestClient

def test_list_users():
    app = create_app(router, dependency_overrides={get_db: get_test_db})
    with TestClient(app) as client:
        resp = client.get("/api/users")
        assert resp.status_code == 200

python

class TestDependencyOverrides:
    """Dependency overrides: original factory replaced with test double."""

    @pytest.mark.anyio
    async def test_override_replaces_factory(self, client_for):
        def get_conn(request):
            return "production_db"

        def test_conn(request):
            return "test_db"

        router = Router()

        @router.get("/data", deps={"conn": get_conn})
        async def data(request, conn=None):
            return JSONResponse({"conn": conn})

        # Without override
        app_no_override = create_app(router)
        async with client_for(app_no_override) as client:
            resp = await client.get("/data")
        assert resp.json()["conn"] == "production_db"

        # With override
        app_with_override = create_app(
            router, dependency_overrides={get_conn: test_conn},
        )
        async with client_for(app_with_override) as client:
            resp = await client.get("/data")
        assert resp.json()["conn"] == "test_db"

    @pytest.mark.anyio
    async def test_override_generator_with_plain_factory(self, client_for):
        """Override a generator dep with a simple factory."""
        def get_conn(request):
            yield "production_conn"

        def test_conn(request):
            return "in_memory_conn"

        router = Router()

        @router.get("/data", deps={"conn": get_conn})
        async def data(request, conn=None):
            return JSONResponse({"conn": conn})

        app = create_app(
            router, dependency_overrides={get_conn: test_conn},
        )
        async with client_for(app) as client:
            resp = await client.get("/data")
        assert resp.json()["conn"] == "in_memory_conn"

    @pytest.mark.anyio
    async def test_override_scoped_to_app_instance(self, client_for):
        """Overrides don't leak between app instances."""
        def get_conn(request):
            return "production"

        def test_conn(request):
            return "test"

        router = Router()

        @router.get("/data", deps={"conn": get_conn})
        async def data(request, conn=None):
            return JSONResponse({"conn": conn})

        # App 1: with override
        app1 = create_app(router, dependency_overrides={get_conn: test_conn})
        # App 2: no override
        app2 = create_app(router)

        async with client_for(app1) as client1:
            resp1 = await client1.get("/data")
        async with client_for(app2) as client2:
            resp2 = await client2.get("/data")

        assert resp1.json()["conn"] == "test"
        assert resp2.json()["conn"] == "production"

    @pytest.mark.anyio
    async def test_override_with_router_level_deps(self, client_for):
        """Overrides work for deps declared at the router level."""
        def get_user(request):
            return "real_user"

        def fake_user(request):
            return "test_user"

        sub = Router()

        @sub.get("/me")
        async def me(request, user=None):
            return JSONResponse({"user": user})

        main = Router()
        main.include_router(sub, prefix="/api", deps={"user": get_user})

        app = create_app(
            main, dependency_overrides={get_user: fake_user},
        )
        async with client_for(app) as client:
            resp = await client.get("/api/me")
        assert resp.json()["user"] == "test_user"

#Error handling during resolution

If a factory raises an exception during resolution, all generator cleanups that have already yielded are run before the exception propagates. For example, if 5 generators have yielded and the 6th factory raises, all 5 are cleaned up in reverse order. This ensures resources are not leaked even when resolution fails partway through.

#Complete example

python

from fastware import Router, create_app, serve

# -- Factories ---------------------------------------------------------------

async def get_db(request):
    conn = await asyncpg.connect(DATABASE_URL)
    try:
        yield conn
    finally:
        await conn.close()

async def get_current_user(request):
    token = request.header("Authorization", "").removeprefix("Bearer ")
    if not token:
        raise HTTPError(401, "Missing token")
    return await verify_token(token)

# -- Routes ------------------------------------------------------------------

router = Router()

@router.get("/profile", deps={"user": get_current_user, "db": get_db})
async def get_profile(request, user, db):
    profile = await db.fetchrow(
        "SELECT * FROM profiles WHERE user_id = $1", user["id"]
    )
    return dict(profile)

@router.get("/health")
async def health(request):
    return {"status": "ok"}

# -- App ---------------------------------------------------------------------

app = create_app(router)

if __name__ == "__main__":
    serve(app, foreground=True, host="127.0.0.1", port=8000)

#API reference

See the full DependencyResolver class reference below, which documents all public methods for factory registration, per-request resolution with identity-based caching, generator cleanup ordering, and dependency override management for test environments:

#`src.fastware.di`

Dependency injection container providing per-request resolution with automatic caching, generator cleanup, and scope-aware dependency override support.

Supports sync/async factory callables and sync/async generator factories (yield pattern). Generator factories get cleanup after the handler returns. Results are cached per-request: the same factory called twice returns the same resolved instance.

#`_request_call_mode`

python

def _request_call_mode(factory: Callable) -> str | None

Determine how factory should receive the request, by introspection.

Returns:

- "positional" if the factory has a parameter that can accept the

request positionally (positional-only, positional-or-keyword, or *args). The request is passed as factory(request).

- "keyword" if the factory has a keyword-only parameter literally

named request. The request is passed as factory(request=request).

- None if the factory takes no request slot (e.g. def f() or

def f(*, x=1)). The factory is called with no arguments.

This prevents wrongly calling factory(request) on a factory whose only parameters are keyword-only (which raises TypeError).

#`DependencyResolver`

Resolves a dict of {name: factory} into {name: value} per request.

overrides maps an original factory to a replacement factory. When an override exists for a factory, the replacement is called instead.

#`resolve`

python

async def resolve(self, deps: dict[str, Callable], request: Any) -> tuple[dict[str, Any], list[tuple[str, Any]]]

Resolve deps against request.

Returns (resolved, cleanups) where resolved is a {name: value} dict and cleanups is a list of ("sync" | "async", generator) pairs to pass to :meth:cleanup.

If a factory raises during resolution, any generator cleanups accumulated so far are run before the exception propagates.

#`cleanup`

python

async def cleanup(cleanups: list[tuple[str, Any]]) -> None

Run generator cleanups in reverse order.

Each generator dependency must yield exactly once. If a generator yields a second time during cleanup (a multi-yield generator, like a FastAPI-style dependency written with two yield statements), that is a programming error: :class:RuntimeError is raised after all cleanups have run.

Errors raised inside a generator's cleanup code (e.g. in a finally block) are logged at ERROR level and suppressed -- a failing cleanup must not mask the handler's response or exception. They are never swallowed silently.

#Dependency Injection

#What DI solves

#Basic usage

#1. Define factory functions

#2. Declare dependencies on routes

#3. Create the app

#Sync and async factories

#Generator factories (yield pattern)

#Sync generator

#Async generator

#Per-request caching

#Router-level dependencies

#Dependency overrides for testing

#Error handling during resolution

#Complete example

#API reference

#src.fastware.di

#_request_call_mode

#DependencyResolver

#resolve

#cleanup

#`src.fastware.di`

#`_request_call_mode`

#`DependencyResolver`

#`resolve`

#`cleanup`