如何创建可以接受 Form 或 JSON 正文的 FastAPI 端点？

Question

我想在 FastAPI 中创建一个可能接收（多部分）Form数据或JSON正文的端点。有没有办法让这样的端点接受或者检测正在接收哪种类型的数据？

Answer 1

选项1

\n

您可以通过依赖函数来做到这一点，在该函数中您检查请求标头的值Content-Type并相应地使用 Starlette 的方法解析主体。请注意，仅仅因为请求的Content-Type标头显示，例如 、application/json或application/x-www-form-urlencoded，multipart/form-data并不总是意味着这是真的，或者传入的数据是有效的 JSON、文件和/或表单数据。因此，try-except在解析正文时，您应该使用块来捕获任何潜在的错误。此外，您可能需要实施各种检查，以确保获得正确类型的数据以及您期望需要的所有字段。对于 JSON 主体，您可以创建一个BaseModel并使用 Pydantic 的parse_obj函数来验证接收到的字典（类似于此答案的方法 3 ）。

\n

关于文件/表单数据，您可以直接使用Starlette的Request对象，更具体地说，是request.form()解析正文的方法，该方法将返回一个包含文件上传和文本输入的FormData不可变多字典（即ImmutableMultiDict）对象。当您发送某些输入的值或 的列表时，您可以使用 multidict 的方法来检索. 对于文件，这将返回一个对象，您可以按照与此答案和此答案相同的方式使用该对象来循环访问文件并检索其内容。您还可以直接从 读取请求正文并使用库解析它，而不是使用 ，如本答案中所示。listformfilesgetlist()listlistUploadFilerequest.form()streamstreaming-form-data

\n

工作示例

\n

from fastapi import FastAPI, Depends, Request, HTTPException\nfrom starlette.datastructures import FormData\nfrom json import JSONDecodeError\n\napp = FastAPI()\n\nasync def get_body(request: Request):\n    content_type = request.headers.get(\'Content-Type\')\n    if content_type is None:\n        raise HTTPException(status_code=400, detail=\'No Content-Type provided!\')\n    elif content_type == \'application/json\':\n        try:\n            return await request.json()\n        except JSONDecodeError:\n            raise HTTPException(status_code=400, detail=\'Invalid JSON data\')\n    elif (content_type == \'application/x-www-form-urlencoded\' or\n          content_type.startswith(\'multipart/form-data\')):\n        try:\n            return await request.form()\n        except Exception:\n            raise HTTPException(status_code=400, detail=\'Invalid Form data\')\n    else:\n        raise HTTPException(status_code=400, detail=\'Content-Type not supported!\')\n\n@app.post(\'/\')\ndef main(body = Depends(get_body)):\n    if isinstance(body, dict):  # if JSON data received\n        return body\n    elif isinstance(body, FormData):  # if Form/File data received\n        msg = body.get(\'msg\')\n        items = body.getlist(\'items\')\n        files = body.getlist(\'files\')  # returns a list of UploadFile objects\n        if files:\n            print(files[0].file.read(10))\n        return msg\n

\n

选项2

\n

另一种选择是使用单个端点，并将文件和/或表单数据参数定义为Optional（查看此答案和此答案，了解如何执行此操作的所有可用方法）。一旦客户端的请求进入端点，您就可以检查定义的参数是否有任何值传递给它们，这意味着它们被客户端包含在请求正文中，并且这是一个具有或的请求Content-Type（application/x-www-form-urlencoded请multipart/form-data注意，如果如果您希望收到任意文件或表单数据，则应该使用上面的选项 1）。否则，如果每个定义的参数仍然是None（意味着客户端没有在请求正文中包含任何参数），那么这可能是一个 JSON 请求，因此，通过尝试将请求正文解析为 JSON 来继续确认。

\n

工作示例

\n

from fastapi import FastAPI, UploadFile, File, Form, Request, HTTPException\nfrom typing import Optional, List\nfrom json import JSONDecodeError\n\napp = FastAPI()\n\n@app.post(\'/\')\nasync def submit(request: Request, items: Optional[List[str]] = Form(None),\n                    files: Optional[List[UploadFile]] = File(None)):\n    # if File(s) and/or form-data were received\n    if items or files:\n        filenames = None\n        if files:\n            filenames = [f.filename for f in files]\n        return {\'File(s)/form-data\': {\'items\': items, \'filenames\': filenames}}\n    else:  # check if JSON data were received\n        try:\n            data = await request.json()\n            return {\'JSON\': data}\n        except JSONDecodeError:\n            raise HTTPException(status_code=400, detail=\'Invalid JSON data\')\n

\n

选项3

\n

另一种选择是定义两个单独的端点；一个用于处理 JSON 请求，另一个用于处理文件/表单数据请求。使用中间件，您可以检查传入请求是否指向您希望用户发送 JSON 或文件/表单数据的路由（在下面的示例中为/路由），如果是，请检查Content-Type与上一个选项类似的选项并相应地将请求重新路由到/submitJSON或/submitForm端点（您可以通过修改path中的属性来实现request.scope）。这种方法的优点是，它允许您像往常一样定义端点，而不必担心如果请求中缺少必填字段或接收到的数据不符合预期格式时处理错误。

\n

工作示例

\n

from fastapi import FastAPI, Request, Form, File, UploadFile\nfrom fastapi.responses import JSONResponse\nfrom typing import List, Optional\nfrom pydantic import BaseModel\n\napp = FastAPI()\n\nclass Item(BaseModel):\n    items: List[str]\n    msg: str\n\n@app.middleware("http")\nasync def some_middleware(request: Request, call_next):\n    if request.url.path == \'/\':\n        content_type = request.headers.get(\'Content-Type\')\n        if content_type is None:\n            return JSONResponse(\n                content={\'detail\': \'No Content-Type provided!\'}, status_code=400)\n        elif content_type == \'application/json\':\n            request.scope[\'path\'] = \'/submitJSON\'\n        elif (content_type == \'application/x-www-form-urlencoded\' or\n              content_type.startswith(\'multipart/form-data\')):\n            request.scope[\'path\'] = \'/submitForm\'\n        else:\n            return JSONResponse(\n                content={\'detail\': \'Content-Type not supported!\'}, status_code=400)\n\n    return await call_next(request)\n\n@app.post(\'/\')\ndef main():\n    return\n\n@app.post(\'/submitJSON\')\ndef submit_json(item: Item):\n    return item\n\n@app.post(\'/submitForm\')\ndef submit_form(msg: str = Form(...), items: List[str] = Form(...),\n                    files: Optional[List[UploadFile]] = File(None)):\n    return msg\n

\n

选项4

\n

我还建议您查看此答案，它提供了有关如何在同一请求中一起发送 JSON 正文和文件/表单数据的解决方案，这可能会让您对要解决的问题有不同的看法。例如，将各种端点的参数声明为Optional并检查哪些参数已从客户端的请求 xe2\x80\x94 中收到，哪些尚未收到，以及使用 Pydantic 的方法model_validate_json()解析 JSON传入 a 的字符串Form可能是解决该问题的另一种方法。请参阅上面链接的答案以获取更多详细信息和示例。

\n

使用 Python 请求测试选项 1、2 和 3

\n

测试.py

\n

import requests\n\nurl = \'http://127.0.0.1:8000/\'\nfiles = [(\'files\', open(\'a.txt\', \'rb\')), (\'files\', open(\'b.txt\', \'rb\'))]\npayload ={\'items\': [\'foo\', \'bar\'], \'msg\': \'Hello!\'}\n \n# Send Form data and files\nr = requests.post(url, data=payload, files=files)  \nprint(r.text)\n\n# Send Form data only\nr = requests.post(url, data=payload)              \nprint(r.text)\n\n# Send JSON data\nr = requests.post(url, json=payload)              \nprint(r.text)\n

\n