`@inject` isn't woking at FastAPI

[fbuccioni]

Hi, I'm trying to implement the dependency injection in FastAPI routes but I didn't make it work, here is a Reproducible Minimal Example.

from typing import Annotated
from dependency_injector import containers, providers
from dependency_injector.providers import Factory
from dependency_injector.wiring import Provide, inject
from fastapi import FastAPI, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

# Service
class ChatService:
    """Minimal chat service with dependencies"""
    def __init__(
        self,
        user: str,
        predefined_param: int
    ):
        self.user = user
    
    async def send_message(self):
        yield "message"

# Contaienr
class ChatContainer(containers.DeclarativeContainer):
    # Chat service factory with dependencies
    chat_service = providers.Factory(
        ChatService, predefined_param=10
    )


# Input model
# Pydantic model for request
class ChatMessage(BaseModel):
    content: str


# FastAPI magic
app = FastAPI(title="Minimal DI Example")

@app.post("/chat/stream")
@inject
async def stream_chat(
    message: ChatMessage,
    chat_service_factory: Annotated[
        Factory[ChatService],
        Depends(Provide[ChatContainer.chat_service.provider])
    ]
):
    chat_service = chat_service_factory(
        user="TestUser",
    )
    
    return StreamingResponse(
        chat_service.send_message(),
        media_type="text/plain"
    )

# RUN!!
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

And it is tested with

curl -v http://0.0.0.0:8000/chat/stream -H 'Content-type: application/json' -d '{"content":"hola"}'

I can do a workaround by using

 chat_service_factory:  Factory[ChatService] = Depends(lambda: ChatContainer.chat_service)

But I want to use the proper types.

[ZipFile]

You forgot an important part, wiring:

@@ -57,4 +57,6 @@
 # RUN!!
 if __name__ == "__main__":
     import uvicorn
+    container = ChatContainer()
+    container.wire([__name__])
     uvicorn.run(app, host="0.0.0.0", port=8000)

Also, you can use Provider marker to save on writing .provider every time:

@@ -1,7 +1,7 @@
 from typing import Annotated
 from dependency_injector import containers, providers
 from dependency_injector.providers import Factory
-from dependency_injector.wiring import Provide, inject
+from dependency_injector.wiring import Provider, inject
 from fastapi import FastAPI, Depends
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
@@ -42,7 +42,7 @@
         message: ChatMessage,
         chat_service_factory: Annotated[
             Factory[ChatService],
-            Depends(Provide[ChatContainer.chat_service.provider])
+            Depends(Provider[ChatContainer.chat_service])
         ]
 ):
     chat_service = chat_service_factory(

[fbuccioni]

Hi @ZipFile , when I use wire it works, but when I separate the files, it fails again :(. The MRE files are below

di_mre.py

from di_mre_src.chat_container import ChatContainer
from di_mre_src.chat_routes import router


# RUN!!
if __name__ == "__main__":
    import uvicorn
    app = FastAPI(title="Minimal DI Example")
    app.include_router(router)

    chat_container = ChatContainer()
    chat_container.wire([__name__])
    
    uvicorn.run(app, host="0.0.0.0", port=8000)

di_mre_src/chat_service.py

class ChatService:
    """Minimal chat service with dependencies"""
    def __init__(
        self,
        user: str,
        predefined_param: int
    ):
        self.user = user
    
    async def send_message(self):
        yield "message"

di_mre_src/chat_container.py

from dependency_injector import containers, providers
from di_mre_src.chat_service import ChatService

class ChatContainer(containers.DeclarativeContainer):
    # Chat service factory with dependencies
    chat_service = providers.Factory(
        ChatService, predefined_param=10
    )

di_mre_src/chat_message_model.py

from pydantic import BaseModel

class ChatMessage(BaseModel):
    content: str

di_mre_src/chat_routes.py

from typing import Annotated
from dependency_injector.providers import Factory
from dependency_injector.wiring import Provider, inject
from fastapi import Depends, APIRouter
from fastapi.responses import StreamingResponse

from di_mre_src.chat_message_model import ChatMessage
from di_mre_src.chat_service import ChatService
from di_mre_src.chat_container import ChatContainer


router = APIRouter(prefix='/chat', tags=["Chat"])

@router.post('/stream')
@inject
async def stream_chat(
    message: ChatMessage,
    chat_service_factory: Annotated[Factory[ChatService], Depends(Provider[ChatContainer.chat_service])]
):
    chat_service = chat_service_factory(
        user="TestUser",
    )
    
    return StreamingResponse(
        chat_service.send_message(),
        media_type="text/plain"
    )

[ZipFile]

You have to provide list of modules/packages to wire. For self-contained example from the OP post container.wire([__name__]) is more than enough, since __name__ resolves to the current and only module. Either list all modules with @inject, or provide entire package to wire recursively: container.wire(packages=["di_mre_src"]). See docs for more details and examples.

[fbuccioni]

Hi @ZipFile, I understand my problem, I didn't notice the manual wiring, I am currently working in a refactor of a large codebase in the transition from monolith to hexagonal, so is hard right now to wire manually.

Then I have to find a way to automate the wiring process so I wrote the following code, I want to know your opinion, if you want to include in the code please give me code guidelines to build a PR if you want.

Here is the code, anyway thanks a lot for the awesome work.

bound_injections.py

import inspect
from dependency_injector.containers import Container
from dependency_injector.wiring import inject
from typing import TypeVar, Callable, Any


__BOUND_CONTAINERS__ = set()

F = TypeVar("F", bound=Callable[..., Any])


def bound_inject(*ContainerClasses: type[Container]) -> Callable[[F], F]:
    module = inspect.getmodule(
        inspect.stack()[1][0]
    )

    if module is None:
        raise RuntimeError("Cannot found caller module")

    for ContainerClass in ContainerClasses:
        if module not in __BOUND_CONTAINERS__:
            __BOUND_CONTAINERS__.add(ContainerClass)

        ContainerClass.wiring_config.modules.append(module)

    return inject


def wire_bound_containers():
    for Container in __BOUND_CONTAINERS__:
        Container().wire()

Implementation:

@@ -1,5 +1,6 @@
 from fastapi import FastAPI

+from bound_injections import wire_bound_containers
 from di_mre_src.chat_container import ChatContainer
 from di_mre_src.chat_routes import router

@@ -9,5 +10,5 @@
     import uvicorn
     app = FastAPI(title="Minimal DI Example")
     app.include_router(router)
-
+    wire_bound_containers()
     uvicorn.run(app, host="0.0.0.0", port=8000)

@@ -4,6 +4,7 @@
 from fastapi import Depends, APIRouter
 from fastapi.responses import StreamingResponse

+from bound_injection import bound_inject
 from di_mre_src.chat_message_model import ChatMessage
 from di_mre_src.chat_service import ChatService
 from di_mre_src.chat_container import ChatContainer
@@ -12,7 +13,7 @@
 router = APIRouter(prefix='/chat', tags=["Chat"])

 @router.post('/stream')
-@inject
+@bound_inject(ChatContainer)
 async def stream_chat(
     message: ChatMessage,
     # chat_service_factory: Factory[ChatService] = Depends(Provide[ChatContainer.chat_service.provider])