如何在单独的文件中使用 FastAPI Depends for endpoint/route？
How to use FastAPI Depends for endpoint/route in separate file?
我在单独的文件中定义了一个 Websocket 端点，例如：
from starlette.endpoints import WebSocketEndpoint
from connection_service import ConnectionService
class WSEndpoint(WebSocketEndpoint):
"""Handles Websocket connections"""
async def on_connect(self,
websocket: WebSocket,
connectionService: ConnectionService = Depends(ConnectionService)):
"""Handles new connection"""
self.connectionService = connectionService
...
并在 main.py 中我将端点注册为：
from fastapi import FastAPI
from starlette.routing import WebSocketRoute
from ws_endpoint import WSEndpoint
app = FastAPI(routes=[ WebSocketRoute("/ws", WSEndpoint) ])
但是我的端点的 Depends 从未得到解决。有什么办法让它起作用吗？
另外，FastAPI 中这种机制的目的是什么？我们不能只使用 local/global 个变量吗？
TL;DR
文档似乎暗示你只能使用Depends来请求函数。
说明
我在 FastAPI 存储库中找到了一个相关的 issue #2057，似乎 Depends(...) 只适用于 requests 而不是其他任何东西。
我通过
确认了这一点
from fastapi import Depends, FastAPI
app = FastAPI()
async def foo_func():
return "This is from foo"
async def test_depends(foo: str = Depends(foo_func)):
return foo
@app.get("/")
async def read_items():
depends_result = await test_depends()
return depends_result
在这种情况下，依赖关系没有得到解决。
针对你的情况，你可以像这样解决依赖关系，
from starlette.endpoints import WebSocketEndpoint
from connection_service import ConnectionService
class WSEndpoint(WebSocketEndpoint):
async def on_connect(
self,
websocket: WebSocket,
<b>connectionService=None</b>
):
<b>if connectionService is None:
connectionService = ConnectionService() # calling the depend function
self.connectionService = connectionService</b>
在学习依赖注入和 routes/endpoints FastAPI 数小时后，我发现了这里。
路由与端点
首先要指出的是，Endpoint是Starlette中存在的概念，FastAPI中没有。在我的问题中，我显示了我使用 WebSocketEndpoint class 的代码，并且依赖注入在 FastAPI 中不起作用。进一步阅读以了解原因。
依赖注入 (DI)
FastAPI 中的 DI 不是我们所知道的 classic 模式，它并没有神奇地解决所有地方的所有依赖关系。
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 可用于解析 classes 的实例，但这不是它的主要目的 + 它在 Python 中没有意义，因为您可以只使用 关闭。 Depends 大放异彩的地方是当您需要某种请求验证时（Django 使用装饰器完成的）。在这种用法中 Depends 很棒，因为它解决了 route 依赖项和那些子依赖项。查看下面的代码，我使用 auth_check.
代码示例
作为奖励，我希望将 websocket 路由作为 class 在单独的文件中使用单独的连接、断开和接收方法。另外，我想在单独的文件中进行身份验证检查，以便能够轻松地交换它。
# 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
我遇到了同样的问题。 Depends/Query 不工作。我停止使用 WebSocketEndpoint 并尝试了这样的事情
socket.py
client_id 将从前端作为令牌查询字符串
# @app.websocket("/ws/hello/token")
async def websocket_hello_endpoint_with_token(websocket: WebSocket, client_id: str = Query(..., alias="token")):
#on_connect
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
#on_receive
await websocket.send_text(f"Token: {client_id} & Message text was: {data}")
except WebSocketDisconnect:
#on_disconnect
pass
main.py
使用websocket_hello_endpoint_with_token
app = FastAPI()
app.add_api_websocket_route("/ws/hello/token", socket.websocket_hello_endpoint_with_token)
客户端
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<form action="" onsubmit="sendMessage(event)">
<label>Token: <input type="text" id="token" autocomplete="off" value="some-key-token"/></label>
<button onclick="connect(event)">Connect</button>
<hr>
<label>Message: <input type="text" id="messageText" autocomplete="off"/></label>
<button>Send</button>
</form>
<ul id='messages'>
</ul>
<script>
var ws = null;
function connect(event) {
var token = document.getElementById("token")
ws = new WebSocket("ws://localhost:6003/ws/hello/token?token=" + token.value);
ws.onopen = function () {
console.log('socket opened');
};
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
<!-- var data = document.createTextNode(event.data) -->
<!-- var content = "message:" + data.message -->
message.appendChild(content)
messages.appendChild(message)
};
ws.onclose = function(e) {
console.log('socket closed from server');
}
ws.onerror = function(err) {
console.error(err)
};
event.preventDefault()
}
function sendMessage(event) {
var input = document.getElementById("messageText")
ws.send(input.value)
input.value = ''
event.preventDefault()
}
</script>
</body>
</html>
我在单独的文件中定义了一个 Websocket 端点，例如：
from starlette.endpoints import WebSocketEndpoint
from connection_service import ConnectionService
class WSEndpoint(WebSocketEndpoint):
"""Handles Websocket connections"""
async def on_connect(self,
websocket: WebSocket,
connectionService: ConnectionService = Depends(ConnectionService)):
"""Handles new connection"""
self.connectionService = connectionService
...
并在 main.py 中我将端点注册为：
from fastapi import FastAPI
from starlette.routing import WebSocketRoute
from ws_endpoint import WSEndpoint
app = FastAPI(routes=[ WebSocketRoute("/ws", WSEndpoint) ])
但是我的端点的 Depends 从未得到解决。有什么办法让它起作用吗？
另外，FastAPI 中这种机制的目的是什么？我们不能只使用 local/global 个变量吗？
TL;DR
文档似乎暗示你只能使用Depends来请求函数。
说明
我在 FastAPI 存储库中找到了一个相关的 issue #2057，似乎 Depends(...) 只适用于 requests 而不是其他任何东西。
我通过
确认了这一点from fastapi import Depends, FastAPI
app = FastAPI()
async def foo_func():
return "This is from foo"
async def test_depends(foo: str = Depends(foo_func)):
return foo
@app.get("/")
async def read_items():
depends_result = await test_depends()
return depends_result
在这种情况下，依赖关系没有得到解决。
针对你的情况，你可以像这样解决依赖关系，
from starlette.endpoints import WebSocketEndpoint
from connection_service import ConnectionService
class WSEndpoint(WebSocketEndpoint):
async def on_connect(
self,
websocket: WebSocket,
<b>connectionService=None</b>
):
<b>if connectionService is None:
connectionService = ConnectionService() # calling the depend function
self.connectionService = connectionService</b>
在学习依赖注入和 routes/endpoints FastAPI 数小时后，我发现了这里。
路由与端点
首先要指出的是，Endpoint是Starlette中存在的概念，FastAPI中没有。在我的问题中，我显示了我使用 WebSocketEndpoint class 的代码，并且依赖注入在 FastAPI 中不起作用。进一步阅读以了解原因。
依赖注入 (DI)
FastAPI 中的 DI 不是我们所知道的 classic 模式，它并没有神奇地解决所有地方的所有依赖关系。
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 可用于解析 classes 的实例，但这不是它的主要目的 + 它在 Python 中没有意义，因为您可以只使用 关闭。 Depends 大放异彩的地方是当您需要某种请求验证时（Django 使用装饰器完成的）。在这种用法中 Depends 很棒，因为它解决了 route 依赖项和那些子依赖项。查看下面的代码，我使用 auth_check.
代码示例
作为奖励，我希望将 websocket 路由作为 class 在单独的文件中使用单独的连接、断开和接收方法。另外，我想在单独的文件中进行身份验证检查，以便能够轻松地交换它。
# 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
我遇到了同样的问题。 Depends/Query 不工作。我停止使用 WebSocketEndpoint 并尝试了这样的事情
socket.py
client_id 将从前端作为令牌查询字符串
# @app.websocket("/ws/hello/token")
async def websocket_hello_endpoint_with_token(websocket: WebSocket, client_id: str = Query(..., alias="token")):
#on_connect
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
#on_receive
await websocket.send_text(f"Token: {client_id} & Message text was: {data}")
except WebSocketDisconnect:
#on_disconnect
pass
main.py
使用websocket_hello_endpoint_with_token
app = FastAPI()
app.add_api_websocket_route("/ws/hello/token", socket.websocket_hello_endpoint_with_token)
客户端
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<form action="" onsubmit="sendMessage(event)">
<label>Token: <input type="text" id="token" autocomplete="off" value="some-key-token"/></label>
<button onclick="connect(event)">Connect</button>
<hr>
<label>Message: <input type="text" id="messageText" autocomplete="off"/></label>
<button>Send</button>
</form>
<ul id='messages'>
</ul>
<script>
var ws = null;
function connect(event) {
var token = document.getElementById("token")
ws = new WebSocket("ws://localhost:6003/ws/hello/token?token=" + token.value);
ws.onopen = function () {
console.log('socket opened');
};
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
<!-- var data = document.createTextNode(event.data) -->
<!-- var content = "message:" + data.message -->
message.appendChild(content)
messages.appendChild(message)
};
ws.onclose = function(e) {
console.log('socket closed from server');
}
ws.onerror = function(err) {
console.error(err)
};
event.preventDefault()
}
function sendMessage(event) {
var input = document.getElementById("messageText")
ws.send(input.value)
input.value = ''
event.preventDefault()
}
</script>
</body>
</html>