How to inject dependencies to my custom function in FastAPI

Viewed 212

I have a custom class that I use for all websocket routes. It looks kind of like this:

class Handler:
   async def __call__(websocket):
      await websocket.accept()
      await self.handle_websocket(websocket)

   @abc.abstractmethod
   async def handle_websocket(websocket):
      pass

So now I have problem: if I want to use dependency in handle_websocket function how do I do that? I'd like to have to be able to do something like this:

async def __call__(websocket):
      await websocket.accept()
      container = get_dependency_injection_container()
      # here I need to pass websocket as an argument. Other arguments in handle_websocket function should be injected
      await container.call(self.handle_websocket, websocket=websocket) 

Any ideas?

1 Answers

I'm not sure if it is possible to do exactly as you suggest, because to solve dependencies, they must be listed in the signature of the path operation (at least this is the only standard way).

I would suggest that you slightly change the approach and use classes as a dependency, getting all the necessary dependencies in the __init__ method.

Demonstration:

import abc
from fastapi import FastAPI, WebSocket, Depends


app = FastAPI()


def foo_dependency() -> 42:
    return 42


class Handler:
    def __init__(self, websocket: WebSocket):
        self.websocket = websocket

    async def __call__(self):
        await self.websocket.accept()
        return await self.handle_websocket()

    @abc.abstractmethod
    async def handle_websocket(self):
        pass


class ConcreteHandler(Handler):

    def __init__(self, websocket: WebSocket, dep: int = Depends(foo_dependency)):
        super().__init__(websocket)
        self.dep = dep

    async def handle_websocket(self):
        return self.dep


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, handler: ConcreteHandler = Depends()):
    print(await handler())
Related