是否有一种 FastAPI 方法可以在全球范围内访问当前的请求数据？

Posted 2023-02-15

【中文标题】是否有一种 FastAPI 方法可以在全球范围内访问当前的请求数据？

【英文标题】：Is there a FastAPI way to access current Request data globally?

【发布时间】：2019-12-03 21:09:13

【问题描述】：

在 FastAPI 框架内：

虽然请求数据当然可以作为参数传递，但我想知道函数是否可以在不传递参数的情况下访问有关当前请求的信息。

免责声明：我不认为全局访问请求数据是一种好的做法，但我有一个用例，如果能够做到这一点会非常好。

【参考方案1】：

我通常会使用生产者-消费者样式的消息队列来执行此操作。我有一个example repo，展示了我如何使用全局队列将数据从发布请求推送到将其广播给客户端的 WebSocket。

虽然这可能不是您的确切用例，但您应该能够适应它。

它的核心是一个将数据推送到队列的 Notifier 类：

async def push(self, msg: str):
    await self.channel.default_exchange.publish(
        Message(msg.encode("ascii")),
        routing_key=self.queue_name,
    )

在消费者方面，我有一个_notify 函数，它从队列接收消息并通过WebSocket 发送：

async def _notify(self, message: IncomingMessage):
    living_connections = []
    while len(self.connections) > 0:
        websocket = self.connections.pop()
        await websocket.send_text(f"message.body")
        living_connections.append(websocket)
    self.connections = living_connections

【参考方案2】：

您可以从 Starlette 获取/设置request.state 的任意属性。

https://www.starlette.io/requests/#other-state

详细解释和实现请参考以下问题：

https://github.com/tiangolo/fastapi/issues/633

【参考方案3】：

你可以使用starletteRequest

例如：

from starlette.requests import Request
from fastapi import FastApi

app = FastApi()
@app.get('/')
def get(request:Request):
    requests_header = request.headers
    return "Hi"

【讨论】：

在这个例子中，我们将请求对象作为参数传递给我们的 get 回调。我们可以将它作为参数传递给其他函数。但是我问的是能够访问来自另一个模块的任何当前请求而不将其作为参数传递。例如，如果我们创建一个 logging.Formatter，希望将请求中的某些内容作为每个日志消息的标准部分包含在内，我们需要一种方法来访问请求对象而不将其传递给 format 方法。

【参考方案4】：

here 提供的解决方案定义了一个上下文管理器，您可以全局访问它。对于每个请求，您都在提取相关信息（如标头）并将其传递给上下文管理器。

由于 fastapi 是使用 Starlette 构建的，因此您可以使用库 starlette-context。它正在创建一个 context 对象，您可以使用它而不将其作为参数传递。主要的警告是您仍然需要将请求对象传递给您的所有路由。

编辑：在starlette-context==0.3.0 中添加了新的中间件。 Starlette 团队开始不鼓励 (here) 使用他们的 BaseHTTPMiddleware，特别是对于 StreamingResponse/FileResponse 端点。您可能想使用RawContextMiddleware，它也不需要请求对象，但它是实验性的，因为Starlette 中没有用于编写没有接口的自定义中间件的文档。但它似乎起作用了。

来自这个库的示例代码来说明：

import uvicorn
from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.middleware import Middleware

from starlette_context import context, plugins
from starlette_context.middleware import ContextMiddleware

middleware = [
    Middleware(
        ContextMiddleware,
        plugins=(
            plugins.RequestIdPlugin(),
            plugins.CorrelationIdPlugin()
        )
    )
]

app = FastAPI(debug=True, middleware=middleware)


@app.route('/')
async def index(request: Request):  # This argument is still needed here
    return JSONResponse(context.data)  # Your context data


uvicorn.run(app, host="0.0.0.0")

【讨论】：

嘿，我写了starlette-context！如果您有任何疑问，请不要犹豫，在 GH 上开票。

嘿@TomWojcik，我有一个关于starlette_context 的问题：我正在开发一个FastAPI 应用程序，我必须添加你的包，因为我需要通过http 标头接收一些数据并将一些东西放在一个稍后使用的上下文。该应用程序运行良好，但我有一些测试失败，因为“您没有使用 ContextMiddleware 或者您试图在请求-响应周期之外访问上下文对象”。现在，我使用同一个工厂来生成具有所有正确中间件的 app 对象，并且测试调用包含我需要测试的标头。我不明白为什么会发生错误

嘿@bruno-ripa，请在 GH 上开一张票，但很可能是由于中间件的顺序不正确。