以同步方式使用 FastAPI，如何获取 POST 请求的原始正文？

2024-05-20

在中使用 FastAPIsync, not async模式，我希望能够接收 POST 请求的原始、未更改的正文。

我能找到的所有例子都显示async代码，当我以正常同步方式尝试时，request.body()显示为协程对象。

当我通过发布一些内容来测试它时XML到这个端点，我得到一个500 "Internal Server Error".

from fastapi import FastAPI, Response, Request, Body

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.post("/input")
def input_request(request: Request):
    # how can I access the RAW request body here?  
    body = request.body()

    # do stuff with the body here  

    return Response(content=body, media_type="application/xml")

这对于 FastAPI 来说是不可能的吗？

注意：简化的输入请求如下所示：

POST http://127.0.0.1:1083/input
Content-Type: application/xml

<XML>
    <BODY>TEST</BODY>
</XML>

我无法控制输入请求的发送方式，因为我需要替换现有的 SOAP API。

Using `async def`终点

如果一个对象是协同例程，则需要等待它。FastAPI实际上是底层的Starlette https://fastapi.tiangolo.com/advanced/using-request-directly/, and 星光法 https://www.starlette.io/requests/#body用于返回请求body are async方法（见源码here https://github.com/encode/starlette/blob/212fa46b23be0701a5963cdeff14f05ed352e22a/starlette/requests.py#L231以及）；因此，人们需要await他们（在一个async def终点）。例如：

from fastapi import Request

@app.post("/input")
async def input_request(request: Request):
    return await request.body()

更新 1 - 使用`def`终点

或者，如果您确信传入的数据是有效的JSON，您可以定义您的端点def相反，并使用Body https://fastapi.tiangolo.com/tutorial/body-fields/字段，如下所示（有关如何发布的更多选项JSON数据，参见这个答案 https://stackoverflow.com/a/70636163/17865804):

from fastapi import Body

@app.post("/input")
def input_request(payload: dict = Body(...)):
    return payload

但是，如果传入数据位于XML格式，如您提供的示例中所示，一种选择是使用传递它们Files https://fastapi.tiangolo.com/tutorial/request-files/相反，如下所示 - 只要您可以控制客户端数据发送到服务器的方式（看看here https://stackoverflow.com/a/70657621/17865804以及）。例子：

from fastapi import File

@app.post("/input") 
def input_request(contents: bytes = File(...)): 
    return contents

更新 2 - 使用`def`终点和`async`依赖性

如中所述这个帖子 https://github.com/tiangolo/fastapi/issues/2574#issuecomment-751787209，你可以使用async 依赖性 https://fastapi.tiangolo.com/tutorial/dependencies/函数来拉出body来自请求。您可以使用async依赖于non-async (i.e., def) 端点也是如此。因此，如果此端点中存在某种阻止您使用的阻止代码async/await https://fastapi.tiangolo.com/async/——我猜这可能是你的情况的原因——这就是要走的路。

注意：我还应该提到这个答案 https://stackoverflow.com/a/71517830/17865804- 这解释了之间的区别def and async def端点（您可能知道）——当您需要使用时还提供解决方案async def（因为你可能需要await对于路线内的协程），但也有一些同步昂贵的 CPU 密集型操作可能会阻塞服务器。请看一看。

下面可以找到前面描述的方法的示例。您可以取消注释time.sleep()行，如果您想确认自己的请求不会阻止其他请求通过，如当您使用正常声明端点时def代替async def，它在外部线程池中运行 https://fastapi.tiangolo.com/async/#path-operation-functions（不管async def依赖函数）。

from fastapi import FastAPI, Depends, Request
import time

app = FastAPI()

async def get_body(request: Request):
    return await request.body()

@app.post("/input")
def input_request(body: bytes = Depends(get_body)):
    print("New request arrived.")
    #time.sleep(5)
    return body

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)