如何注入依赖到我的自定义函数在FastAPI

Question

我有一个用于所有websocket路由的自定义类。它看起来像这样:

class Handler:
async def __call__(websocket):
await websocket.accept()
await self.handle_websocket(websocket)
@abc.abstractmethod
async def handle_websocket(websocket):
pass

所以现在我有问题:如果我想在handle_websocket函数中使用依赖，我该怎么做?我希望能够做这样的事情:

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) 

任何想法?

Answer 1

我不确定是否可以完全按照您的建议做，因为要解决依赖关系，它们必须在路径操作的签名中列出(至少这是唯一的标准方法)。

我建议你稍微改变一下方法，使用类作为依赖项，在__init__方法中获得所有必要的依赖项。

示范:

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())