如何在单独的文件中使用 FastAPI Depends 的端点/路由？

Posted 2023-02-16

【中文标题】如何在单独的文件中使用 FastAPI Depends 的端点/路由？

【英文标题】：How to use FastAPI Depends for endpoint/route in separate file?

【发布时间】：2020-11-11 07:13:45

【问题描述】：

我在单独的文件中定义了一个 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 中这种机制的目的是什么？我们不能只使用局部/全局变量吗？

【参考方案1】：

在学习使用 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

【讨论】：

感谢您的解释。当我听到依赖注入时，我起初对 FastAPI bc 感到困惑，我认为容器和依赖项在任何地方都被注入/解析，但正如您所解释的，依赖注入仅在开始于 FastAPI 路径操作时才有效，基于每个请求。作为一个服务器，这并不是我最初想象的真正的限制，它只是我不习惯的一种模式。

【参考方案2】：

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,
            connectionService=None
    ):
        if connectionService is None:
            connectionService = ConnectionService()  # calling the depend function

        self.connectionService = connectionService

【讨论】：

哦，谢谢，文档里有提到吗？我没有注意那个。所以，是的，你是对的，唯一解决依赖关系的地方是 FastAPI add_api_route 和 add_api_websocket_route，或者如果使用它们的装饰器类似物。您可以在一个文件中定义一个函数，然后在 main 中注册它，它会正常工作。 __ 虽然，我不明白这种“依赖注入”机制在 FastAPI 范围内的用途，但它解决了什么问题？

我认为他们没有提到使用Depends()仅用于请求，但是，我找不到任何显示相反的示例.所以，我得出了上述结论。另外，我尝试过的例子让我相信。

他们没有提到它，但据我所知，这是它现在工作的唯一方式。我可以在fastapi.applications.py 中看到它的源代码，如果在fastpi.dependencies.utils.get_dependant 中放置断点

实际上，您的代码没有多大意义，因为每个连接都会有单独的WSEndpoint 实例，因此您总是会点击is None 检查。另外我想使用ConnectionService作为所有连接的单例，当然我可以用装饰器或其他方法单独实现它。

"...我想将 ConnectionService 用作所有连接的单例" 如果您想让它成为单例，是否可以将其定义为上课？

【参考方案3】：

我也面临同样的问题。依赖/查询不起作用。我停止使用 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

ma​​in.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 端点管理设计解决方案。最后我觉得它看起来很优雅。

@andnik WSRoute 在连接后自动断开

我需要查看您的代码，我们在项目中成功使用了 WSRoute，它运行良好。也许有一些语法错误。尝试调试以找出实际导致断开连接的原因。但如果你想使用我的版本，我可以向你保证它可以工作。