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

Question

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

Answer 1

经过几个小时的学习后，我发现了 FastAPI 中的依赖注入和路由/端点。

路由与端点

首先要指出的是，这是StarletteEndpoint中存在的概念，而FastAPI中没有。在我的问题中，我展示了使用类的代码，并且依赖注入在 FastAPI 中不起作用。进一步阅读以了解原因。WebSocketEndpoint

依赖注入（DI）

FastAPI 中的 DI 并不是我们所知道的经典模式，它并没有神奇地解决所有地方的依赖关系。

Depends仅针对 FastAPI 路由进行解析，这意味着使用方法：add_api_routeand add_api_websocket_route，或其装饰器类似物：api_routeand websocket，它们只是前两个的包装器。

然后，当请求通过 FastAPI 到达路由时，依赖关系将得到解决。了解 FastAPI 正在解决依赖关系而不是 Starlette 非常重要。FastAPI 构建在 Starlette 之上，您可能还想使用一些“原始”Starlette 功能，例如：add_route或add_websocket_route，但是您将无法解决Depends这些问题。

另外，FastAPI 中的 DI 可用于解析类的实例，但这不是其主要目的 + 它在 Python 中没有意义，因为您可以只使用CLOSURE。Depends当你需要某种请求验证时（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

Answer 2

长话短说

该文档似乎暗示您只能用于Depends请求功能。

解释

我在 FastAPI 存储库中发现了一个相关问题 #2057Depends(...) ，它似乎只适用于请求，而不适用于其他任何内容。

我确认了这一点，

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