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

Question

我想在 FastAPI 中创建一个端点，它可以接收（多部分）

Form

数据或

JSON

正文。有没有办法让这样的端点接受其中之一，或者检测正在接收哪种类型的数据？

Answer 1

选项1

您可以通过使用 dependency 函数来做到这一点，您可以在其中检查

Content-Type

请求标头的值，并相应地使用 Starlette 的方法解析正文。请注意，仅仅因为请求的 Content-Type

 标头显示，例如 

application/json

、

application/x-www-form-urlencoded

 或 

multipart/form-data，并不总是意味着这是真的，或者传入的数据是有效的 JSON，或者文件和/或表单数据。因此，在解析正文时，您应该使用 try-except

 块来捕获任何潜在的错误。此外，您可能需要实施各种检查，以确保获得正确类型的数据以及您期望需要的所有字段。对于 JSON 正文，您可以创建一个 

BaseModel

 并使用 Pydantic 的 

parse_obj 函数来验证接收到的字典（类似于 this answer 的方法 3）。

关于文件/表单数据，您可以直接

使用Starlette的Request

对象

，更具体地说，可以使用request.form()方法来解析body，该方法将返回一个不可变的FormData对象multidict（即 ImmutableMultiDict）包含 both 文件上传和文本输入。当您发送某些 list

 输入的 

form

 值或 

files

 列表时，您可以使用多字典的 

getlist() 方法来检索 list

。对于文件，这将返回 

list

对象的

UploadFile

，您可以按照与 thisanswer 和 thisanswer 相同的方式使用这些对象来循环遍历文件并检索其内容。您还可以直接从 request.form()

 读取请求正文并使用 

stream

 库解析它，而不是使用 

streaming-form-data

，如

这个答案中所示。

工作示例

from fastapi import FastAPI, Depends, Request, HTTPException
from starlette.datastructures import FormData
from json import JSONDecodeError

app = FastAPI()

async def get_body(request: Request):
    content_type = request.headers.get('Content-Type')
    if content_type is None:
        raise HTTPException(status_code=400, detail='No Content-Type provided!')
    elif content_type == 'application/json':
        try:
            return await request.json()
        except JSONDecodeError:
            raise HTTPException(status_code=400, detail='Invalid JSON data')
    elif (content_type == 'application/x-www-form-urlencoded' or
          content_type.startswith('multipart/form-data')):
        try:
            return await request.form()
        except Exception:
            raise HTTPException(status_code=400, detail='Invalid Form data')
    else:
        raise HTTPException(status_code=400, detail='Content-Type not supported!')

@app.post('/')
def main(body = Depends(get_body)):
    if isinstance(body, dict):  # if JSON data received
        return body
    elif isinstance(body, FormData):  # if Form/File data received
        msg = body.get('msg')
        items = body.getlist('items')
        files = body.getlist('files')  # returns a list of UploadFile objects
        if files:
            print(files[0].file.read(10))
        return msg

选项2

另一种选择是使用单个端点，并将文件和/或表单数据参数定义为

Optional

（请查看

这个答案和这个答案，了解所有可用的方法要做到这一点）。一旦客户端的请求进入端点，您就可以检查定义的参数是否有任何值传递给它们，这意味着它们被客户端包含在请求正文中，并且这是一个具有 Content-Type

 或 

application/x-www-form-urlencoded

 或 

的请求multipart/form-data

（请注意，如果您希望接收任意文件或表单数据，则应该使用上面的

选项1）。否则，如果每个定义的参数仍然是None

（意味着客户端没有在请求正文中包含任何参数），那么这可能是一个 JSON 请求，因此，通过尝试解析请求正文来继续确认作为 JSON。

工作示例

from fastapi import FastAPI, UploadFile, File, Form, Request, HTTPException
from typing import Optional, List
from json import JSONDecodeError

app = FastAPI()

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

选项3

另一种选择是定义两个单独的端点；一个用于处理 JSON 请求，另一个用于处理文件/表单数据请求。使用

中间件，您可以检查传入请求是否指向您希望用户发送 JSON 或文件/表单数据的路由（在下面的示例中为 /

 路由），如果是，请检查 

Content-Type

 与上一个选项类似，并相应地将请求重新路由到 

/submitJSON

 或 

/submitForm

 端点（您可以通过修改 

path

 中的 

request.scope

 属性来实现这一点，如 

this 答案 中所示）。这种方法的优点是，它允许您照常定义端点，而不必担心如果请求中缺少必填字段或接收到的数据不符合预期格式时处理错误。

工作示例

from fastapi import FastAPI, Request, Form, File, UploadFile
from fastapi.responses import JSONResponse
from typing import List, Optional
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    items: List[str]
    msg: str

@app.middleware("http")
async def some_middleware(request: Request, call_next):
    if request.url.path == '/':
        content_type = request.headers.get('Content-Type')
        if content_type is None:
            return JSONResponse(
                content={'detail': 'No Content-Type provided!'}, status_code=400)
        elif content_type == 'application/json':
            request.scope['path'] = '/submitJSON'
        elif (content_type == 'application/x-www-form-urlencoded' or
              content_type.startswith('multipart/form-data')):
            request.scope['path'] = '/submitForm'
        else:
            return JSONResponse(
                content={'detail': 'Content-Type not supported!'}, status_code=400)

    return await call_next(request)

@app.post('/')
def main():
    return

@app.post('/submitJSON')
def submit_json(item: Item):
    return item

@app.post('/submitForm')
def submit_form(msg: str = Form(...), items: List[str] = Form(...),
                    files: Optional[List[UploadFile]] = File(None)):
    return msg

选项 4

我还建议您查看

这个答案，它提供了如何在同一请求中同时发送 JSON 正文和文件/表单数据的解决方案，这可能会让您对正在尝试的问题有不同的看法来解决。例如，将各种端点的参数声明为 Optional

 并检查哪些已从客户端请求中收到，哪些尚未收到，以及使用 Pydantic 的 

model_validate_json()

 方法来解析在 

Form

 参数中传递的 JSON 字符串——可能是解决问题的另一种方法。请参阅上面链接的答案以获取更多详细信息和示例。

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

测试.py

import requests

url = 'http://127.0.0.1:8000/'
files = [('files', open('a.txt', 'rb')), ('files', open('b.txt', 'rb'))]
payload ={'items': ['foo', 'bar'], 'msg': 'Hello!'}
 
# Send Form data and files
r = requests.post(url, data=payload, files=files)  
print(r.text)

# Send Form data only
r = requests.post(url, data=payload)              
print(r.text)

# Send JSON data
r = requests.post(url, json=payload)              
print(r.text)