Example of DI injection for fastapi

di_adapter.py

import functools
from inspect import signature

from di.container import Container
from di.dependant import Dependant
from di.executors import AsyncExecutor
from fastapi import APIRouter


def copy_signature(source_fct):
    def copy(target_fct):
        sig = signature(source_fct)
        params = {
            k: v
            for k, v in sig.parameters.items()
            if k != "self"
        }
        target_fct.__signature__ = sig.replace(
            parameters=params.values(),
        )
        return target_fct

    return copy


class ControllerAdapter:
    def __init__(
            self,
            router: APIRouter,
            container: Container,
            executor: AsyncExecutor,
    ):
        self.router = router
        self.container = container
        self.executor = executor

    def register(self, controller_class, func, path, **kwargs):
        solved = self.container.solve(
            Dependant(controller_class, scope="request"), scopes=["request"],
        )

        @copy_signature(func)
        @functools.wraps(func)
        async def forwards(*args, **kwargs):
            with self.container.enter_scope("request") as state:
                controller = await self.container.execute_async(
                    solved, executor=self.executor, state=state
                )
                return await func(controller, *args, **kwargs)

        print("self.router.add_api_route", path, forwards, kwargs)
        self.router.add_api_route(path, forwards, **kwargs)
        return forwards

main.py

from dataclasses import dataclass

from di.container import Container, bind_by_type
from di.dependant import Dependant
from di.executors import AsyncExecutor
from fastapi import FastAPI, APIRouter

from di_adapater import ControllerAdapter


class RepoProtocol:
    pass


class Repo:
    pass


@dataclass
class Data:
    title: str


class Controller:
    def __init__(self, repo: RepoProtocol):
        print("init", repo)
        self.repo = repo

    @classmethod
    def register(cls, adapter: ControllerAdapter):
        adapter.register(cls, cls.update, "/", methods=["POST"])

    async def update(self, body: Data) -> Data:
        print(body)
        body.title += " processed"
        return body


router = APIRouter()
container = Container()
container.bind(bind_by_type(Dependant(Repo, scope="request"), RepoProtocol))
executor = AsyncExecutor()
a = ControllerAdapter(router, container, executor)
Controller.register(a)

app = FastAPI()
app.include_router(router)