在学习使用 FastAPI 中的依赖注入和路由/端点数小时后，我发现了。
路由与端点
首先要指出Endpoint是Starlette中存在的概念，FastAPI中没有。在我的问题中，我显示了使用 WebSocketEndpoint 类的代码，并且依赖注入在 FastAPI 中不起作用。进一步阅读以了解原因。
依赖注入 (DI)
FastAPI 中的 DI 不是我们所知道的经典模式，它不能神奇地解决所有地方的所有依赖关系。
Depends 仅适用于 FastAPI 路由，这意味着使用方法：add_api_route 和 add_api_websocket_route，或它们的装饰器类似物：api_route 和 websocket，它们只是前两个的包装器。
然后当请求通过 FastAPI 到达路由时，依赖关系将被解决。了解 FastAPI 解决依赖关系而不是 Starlette 很重要。 FastAPI 是在 Starlette 之上构建的，您可能还想使用一些“原始”的 Starlette 功能，例如：add_route 或 add_websocket_route，但是您将没有 Depends 分辨率 .
此外，FastAPI 中的 DI 可用于解析类的实例，但这不是其主要用途 + 在 Python 中没有意义，因为您可以使用 CLOSURE。 Dependsshine 是当您需要某种请求验证时（Django 使用装饰器完成的）。在这种用法中Depends 很棒，因为它解决了route 依赖项和那些子依赖项。在下面查看我的代码，我使用auth_check。
代码示例
作为奖励，我希望将 websocket 路由作为一个单独的文件中的一个类，并使用单独的连接、断开连接和接收方法。另外，我想在单独的文件中进行身份验证检查，以便能够轻松地将其交换。
# main.py
from fastapi import FastAPI
from ws_route import WSRoute
app = FastAPI()
app.add_api_websocket_route("/ws", WSRoute)
# auth.py
from fastapi import WebSocket
def auth_check(websocket: WebSocket):
# `websocket` instance is resolved automatically
# and other `Depends` as well. They are what's called sub dependencies.
# Implement your authentication logic here:
# Parse Headers or query parameters (which is usually a way for websockets)
# and perform verification
return True
# ws_route.py
import typing
import starlette.status as status
from fastapi import WebSocket, WebSocketDisconnect, Depends
from auth import auth_check
class WSRoute:
def __init__(self,
websocket: WebSocket,
is_authenticated: bool = Depends(auth_check)):
self._websocket = websocket
def __await__(self) -> typing.Generator:
return self.dispatch().__await__()
async def dispatch(self) -> None:
# Websocket lifecycle
await self._on_connect()
close_code: int = status.WS_1000_NORMAL_CLOSURE
try:
while True:
data = await self._websocket.receive_text()
await self._on_receive(data)
except WebSocketDisconnect:
# Handle client normal disconnect here
pass
except Exception as exc:
# Handle other types of errors here
close_code = status.WS_1011_INTERNAL_ERROR
raise exc from None
finally:
await self._on_disconnect(close_code)
async def _on_connect(self):
# Handle your new connection here
await self._websocket.accept()
pass
async def _on_disconnect(self, close_code: int):
# Handle client disconnect here
pass
async def _on_receive(self, msg: typing.Any):
# Handle client messaging here
pass