Advanced DI

Note

It's assumed that you already have a good understanding of dependency injection before reading this page.

FastAPI DI limitations

When dealing with Depends, you may have noticed that the mechanism behind it is essentially just a function call identified by its explicitly referenced name (with associated finalization and caching if specified).

Of course, it's possible to use any callable object, not necessarily a function, but the principle remains the same — the "dependent" code is hardwired to a specific implementation of the dependency. You can think of it as electronic keys 🎫 and doors🚪: to open the right door, you need to find the key that fits exactly (and only) for it.

Let's see how this approach can get confusing with the increasing complexity of our application. Imagine we have two HTTP clients: one communicating with a distant city 🏢 and the other with a nearby village 🏠. The city server determines the IP "under the hood", while the village one requires it to be explicitly passed (and returns it with the main part of the response, even though we didn't ask for it 🤪).

Python 3.8+

import asyncio
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass
from functools import lru_cache
from typing import Annotated

from fastapi import APIRouter, FastAPI
from fastapi.params import Depends
from fastapi.responses import PlainTextResponse


@lru_cache
@dataclass
class ClientSettings:
    timeout: int


@dataclass
class Client(ABC):
    settings: ClientSettings

    async def _connect(self):
        await asyncio.sleep(min(random.randint(1, 5), self.settings.timeout))

    @abstractmethod
    async def request(self) -> str:
        raise NotImplementedError


class CityClient(Client):
    async def request(self) -> str:
        await self._connect()

        return "Response"


class VillageClient(Client):
    @staticmethod
    def _get_ip() -> str:
        return "127.0.0.1"

    async def request(self) -> str:
        ip = self._get_ip()
        await self._connect()

        return f"Response, {ip}"


def get_client_settings() -> ClientSettings:
    return ClientSettings(timeout=3)


ClientSettingsDep = Annotated[ClientSettings, Depends(get_client_settings)]


def get_city_client(settings: ClientSettingsDep) -> CityClient:
    return CityClient(settings=settings)


CityClientDep = Annotated[CityClient, Depends(get_city_client)]


def get_village_client(settings: ClientSettingsDep) -> VillageClient:
    return VillageClient(settings=settings)


VillageClientDep = Annotated[VillageClient, Depends(get_village_client)]


geo_router = APIRouter(prefix="/geo", tags=["Geo"])


@geo_router.get(
    "",
    response_class=PlainTextResponse,
    summary="Fetches geodata.",
)
async def get(client: CityClientDep) -> str:
    return await client.request()


def main() -> FastAPI:
    app = FastAPI(title="DI != DI")

    for router in (geo_router,):
        app.include_router(router, prefix="/api")

    return app

🤓 Other versions and variants

Python 3.8+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

import asyncio
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass
from functools import lru_cache

from fastapi import APIRouter, FastAPI
from fastapi.params import Depends
from fastapi.responses import PlainTextResponse


@lru_cache
@dataclass
class ClientSettings:
    timeout: int


@dataclass
class Client(ABC):
    settings: ClientSettings

    async def _connect(self):
        await asyncio.sleep(min(random.randint(1, 5), self.settings.timeout))

    @abstractmethod
    async def request(self) -> str:
        raise NotImplementedError


class CityClient(Client):
    async def request(self) -> str:
        await self._connect()
        return "Response"


class VillageClient(Client):
    @staticmethod
    def _get_ip() -> str:
        return "127.0.0.1"

    async def request(self) -> str:
        ip = self._get_ip()
        await self._connect()

        return f"Response. Passed ip: {ip}"


def get_client_settings() -> ClientSettings:
    return ClientSettings(timeout=3)


def get_city_client(
    settings: ClientSettings = Depends(get_client_settings),
) -> CityClient:
    return CityClient(settings=settings)


def get_village_client(
    settings: ClientSettings = Depends(get_client_settings),
) -> VillageClient:
    return VillageClient(settings=settings)


geo_router = APIRouter(prefix="/geo", tags=["Geo"])


@geo_router.get(
    "",
    response_class=PlainTextResponse,
    summary="Fetches geodata.",
)
async def get(client: CityClient = Depends(get_city_client)) -> str:
    return await client.request()


def main() -> FastAPI:
    app = FastAPI(title="DI != DI")

    for router in (geo_router,):
        app.include_router(router, prefix="/api")

    return app

Note

Some functions of the auxiliary code used here can be explored on these pages:

• Settings and environment variables for lru_cache.
• Bigger applications — multiple files for APIRouter.
• Custom response - HTML, Stream, File, others for PlainTextResponse.

So, we need:

1. A factory function for each dependency.
2. A construction with Annotated for each dependency to avoid duplicating Depends if one dependency will be used in multiple path operations.
3. A unique name for each construction with Annotated so as not to shadow the origin class name.
4. A lru_cache decorator for each singleton.

While manageable at first, this list grows quickly and spreads across your codebase — making maintenance more difficult.

And the most important, path operations depend on the concrete implementation, not the abstraction — violating the Dependency inversion principle.

We can at least try to reduce the amount of code.

Python 3.8+

import asyncio
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass
from functools import lru_cache
from typing import Annotated

from fastapi import APIRouter, FastAPI
from fastapi.params import Depends
from fastapi.responses import PlainTextResponse


@lru_cache
@dataclass
class ClientSettings:
    timeout: int


def get_client_settings() -> ClientSettings:
    return ClientSettings(timeout=3)


ClientSettingsDep = Annotated[ClientSettings, Depends(get_client_settings)]


@dataclass
class Client(ABC):
    settings: ClientSettingsDep

    async def _connect(self):
        await asyncio.sleep(min(random.randint(1, 5), self.settings.timeout))

    @abstractmethod
    async def request(self) -> str:
        raise NotImplementedError


class CityClient(Client):
    async def request(self) -> str:
        await self._connect()
        return "Response"


class VillageClient(Client):
    @staticmethod
    def _get_ip() -> str:
        return "127.0.0.1"

    async def request(self) -> str:
        ip = self._get_ip()
        await self._connect()

        return f"Response, {ip}"


CityClientDep = Annotated[CityClient, Depends(CityClient)]

VillageClientDep = Annotated[VillageClient, Depends(VillageClient)]


geo_router = APIRouter(prefix="/geo", tags=["Geo"])


@geo_router.get(
    "",
    response_class=PlainTextResponse,
    summary="Fetches geodata.",
)
async def get(client: CityClientDep) -> str:
    return await client.request()


def main() -> FastAPI:
    app = FastAPI(title="DI != DI")

    for router in (geo_router,):
        app.include_router(router, prefix="/api")

    return app

🤓 Other versions and variants

Python 3.8+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

import asyncio
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass
from functools import lru_cache

from fastapi import APIRouter, FastAPI
from fastapi.params import Depends
from fastapi.responses import PlainTextResponse


@lru_cache
@dataclass
class ClientSettings:
    timeout: int


def get_client_settings() -> ClientSettings:
    return ClientSettings(timeout=3)


@dataclass
class Client(ABC):
    settings: ClientSettings = Depends(get_client_settings)

    async def _connect(self):
        await asyncio.sleep(min(random.randint(1, 5), self.settings.timeout))

    @abstractmethod
    async def request(self) -> str:
        raise NotImplementedError


class CityClient(Client):
    async def request(self) -> str:
        await self._connect()
        return "Response"


class VillageClient(Client):
    @staticmethod
    def _get_ip() -> str:
        return "127.0.0.1"

    async def request(self) -> str:
        ip = self._get_ip()
        await self._connect()

        return f"Response, {ip}"


geo_router = APIRouter(prefix="/geo", tags=["Geo"])


@geo_router.get(
    "",
    response_class=PlainTextResponse,
    summary="Fetches geodata.",
)
async def get(client: CityClient = Depends(CityClient)) -> str:
    return await client.request()


def main() -> FastAPI:
    app = FastAPI(title="DI != DI")

    for router in (geo_router,):
        app.include_router(router, prefix="/api")

    return app

But this comes at a cost — the DI logic is now intertwined with the business layer, causing harder decoupling and less flexible testing. And we don't solve original problems.

Dishka integration

Tip

You only need third-party solutions if you want to fully comply with dependency inversion. If you only need dependency injection, it might be better to stay with FastAPI Depends: it nicely does the job it was designed for.

There are many DI libraries in different stages of their lifecycle. As an example, let's consider Dishka because:

• it has strong community support and many integrations, including with FastAPI.
• it builds on lessons learned from earlier DI instruments.
• it complies with DIP through providers and IoC containers.
• it's async- and types-oriented — just like FastAPI.
• it provides an extensive, flexible API ready for complex use cases.

Info

If you're familiar with other frameworks (and not only web), there's good news for you: Dishka has integrations with a large part of them. 😎

Here's how we can rewrite our application.

Python 3.10+

import asyncio
import random
from abc import ABC, abstractmethod
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from dataclasses import dataclass

from dishka import (
    FromDishka,
    Provider,
    Scope,
    from_context,
    make_async_container,
    provide,
)
from dishka.integrations.fastapi import DishkaRoute, setup_dishka
from fastapi import APIRouter, FastAPI
from fastapi.responses import PlainTextResponse


@dataclass
class ClientSettings:
    timeout: int


@dataclass
class Client(ABC):
    settings: ClientSettings

    async def _connect(self):
        await asyncio.sleep(min(random.randint(1, 5), self.settings.timeout))

    @abstractmethod
    async def request(self) -> str:
        raise NotImplementedError


class CityClient(Client):
    async def request(self) -> str:
        await self._connect()

        return "Response"


class VillageClient(Client):
    @staticmethod
    def _get_ip() -> str:
        return "127.0.0.1"

    async def request(self) -> str:
        ip = self._get_ip()
        await self._connect()

        return f"Response, {ip}"


class GeoProvider(Provider):
    scope = Scope.REQUEST

    provides = from_context(provides=ClientSettings, scope=Scope.APP) + provide(
        source=CityClient, provides=Client
    )


geo_router = APIRouter(prefix="/geo", tags=["Geo"], route_class=DishkaRoute)


@geo_router.get(
    "",
    response_class=PlainTextResponse,
    summary="Fetches geodata.",
)
async def get(client: FromDishka[Client]) -> str:
    return await client.request()


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    yield
    await app.state.dishka_container.close()


def main() -> FastAPI:
    app = FastAPI(lifespan=lifespan)

    for router in (geo_router,):
        app.include_router(router, prefix="/api")

    container = make_async_container(
        GeoProvider(), context={ClientSettings: ClientSettings(timeout=3)}
    )
    setup_dishka(container, app)

    return app

Now, the implementation binding is done in configuration, and our path operation simply depends on an interface — not caring how it’s implemented.

Back to our analogy. Now we have just one key 🎫 that we can assign (at the reception desk 😏) to the door 🚪 we need now.

Learn more

Dishka also supports techniques you already know from Depends such as:

• dependencies with yield.
• automatic analysis of __init__.

and introduces new ones:

• context data.
• custom scopes.
• components.

Details — in the excellent documentation.