如何将FastAPI请求转发到另一台服务器？

Question

我有一个用于测试/开发目的的 FastAPI 应用程序。我想要的是，到达我的应用程序的任何请求都会按原样自动发送到另一台服务器上的另一个应用程序，并具有完全相同的参数和相同的端点。这不是重定向，因为我仍然希望应用程序像往常一样处理请求并返回值。我只想向不同服务器上的不同版本的应用程序发起类似的请求，而不需要等待其他服务器的答复，以便其他应用程序获取该请求，就像原始请求发送给它一样。

我怎样才能做到这一点？以下是我用于处理请求的示例代码：

@app.post("/my_endpoint/some_parameters")
def process_request(
    params: MyParamsClass,
    pwd: str = Depends(authenticate),
):
    # send the same request to http://my_other_url/my_endpoint/
    return_value = process_the_request(params)
    return return_value.as_json()

Answer 1

您可以使用库AsyncClient()中的httpx，如此答案、此答案和此答案中所述（请查看这些答案以获取有关下面演示的方法的更多详细信息）。您可以在事件处理程序Client内部生成一个事件处理程序，将其存储在实例 \xe2\x80\x94 中，如此处所述，以及此处和此处\xe2\x80\x94，并在每次需要时重复使用它。一旦完成，您可以使用事件处理程序明确地完成它（更新：由于事件现在已弃用，并且将来可能会完全删除，下面的示例已根据此答案进行更新，该示例演示了如何使用startupappcloseClientshutdownstartupshutdownlifespan而是事件处理程序）。

\n

工作示例

\n

主服务器

\n

当构建即将转发到其他服务器的请求时，主服务器使用从客户端的请求中request.stream()读取请求，它提供了一个迭代器，这样如果客户端发送了一个带有一些大主体的请求（例如例如，客户端上传一个大文件），主服务器在转发请求之前不必等待整个文件被接收并加载到内存中，如果您使用相反的方式，就会发生这种情况，这可能会导致服务器问题，如果整个请求无法放入服务器的 RAM 中。bodyasyncbodyawait request.body()body

\n

您可以按照下面定义的相同方式添加多个路由/upload，指定路径以及端点的 HTTP 方法。请注意，/upload下面的路线使用 Starlette 的path转换器来捕获任意路径，如此处和此处所示。如果您愿意，您还可以指定确切的路径参数，但如果有多个路径参数，下面的示例提供了更方便的方法。无论如何，将根据请求应转发到的其他服务器中的端点来评估路径（如下所示），您可以在其中显式指定路径参数。

\n

from fastapi import FastAPI, Request\nfrom fastapi.responses import StreamingResponse\nfrom starlette.background import BackgroundTask\nfrom contextlib import asynccontextmanager\nimport httpx\n\n\n@asynccontextmanager\nasync def lifespan(app: FastAPI):\n    # Initialise the Client on startup and add it to the state\n    # http://127.0.0.1:8001/ is the base_url of the other server that requests should be forwarded to\n    async with httpx.AsyncClient(base_url=\'http://127.0.0.1:8001/\') as client:\n        yield {\'client\': client}\n        # The Client closes on shutdown \n\n\napp = FastAPI(lifespan=lifespan)\n\n\nasync def _reverse_proxy(request: Request):\n    client = request.state.client\n    url = httpx.URL(path=request.url.path, query=request.url.query.encode(\'utf-8\'))\n    req = client.build_request(\n        request.method, url, headers=request.headers.raw, content=request.stream()\n    )\n    r = await client.send(req, stream=True)\n    return StreamingResponse(\n        r.aiter_raw(),\n        status_code=r.status_code,\n        headers=r.headers,\n        background=BackgroundTask(r.aclose)\n    )\n\n\napp.add_route(\'/upload/{path:path}\', _reverse_proxy, [\'POST\'])\n\n\nif __name__ == \'__main__\':\n    import uvicorn\n    uvicorn.run(app, host=\'0.0.0.0\', port=8000)\n

\n

其他服务器

\n

同样，为了本示例的简单性，该Request对象用于读取请求主体，但您可以像往常一样定义UploadFile、Form、 Pydantic 模型和其他参数/​​依赖项，这对于验证目的也很有用（请参阅此处的相关答案）和这里）。在下面的示例中，服务器正在侦听端口8001。

\n

from fastapi import FastAPI, Request\n\n\napp = FastAPI()\n\n\n@app.post(\'/upload/{p1}/{p2}\')\nasync def upload(p1: str, p2: str, q1: str, request: Request):\n    body = await request.body()\n    print(f\'p1: {p1}, p2: {p2}, q1: {q1}\\nbody: {body}\\n\')\n    return \'OK\'\n    \n    \nif __name__ == \'__main__\':\n    import uvicorn\n    uvicorn.run(app, host=\'0.0.0.0\', port=8001)\n

\n

使用以下命令测试上面的示例httpx

\n

import httpx\n\nurl = \'http://127.0.0.1:8000/upload/hello/world\'\nparams = {\'q1\': \'This is a query param\'}\n\n# send a multipart/form-data encoded request\nfiles = {\'file\': open(\'file.txt\', \'rb\')}\nr = httpx.post(url, params=params, files=files)\nprint(r.content)\n\n# send an application/json encoded request\npayload = {\'msg\': \'This is JSON data\'}\nr = httpx.post(url, params=params, json=payload)\nprint(r.content)\n

\n