如何创建一个可以接受表单或 JSON 正文的 FastAPI 端点？

Question

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

Answer 1

选项1

您可以通过使用dependency 函数来执行此操作，在其中相应地检查Content-Type 请求标头和parse the body using Starlette's methods 的值。请注意，仅仅因为请求的 Content-Type 标头说，例如，application/json 或 application/x-www-form-urlencoded，并不总是意味着这是真的，或者传入的数据是有效的 JSON/Form。因此，您应该在解析正文时使用try-except 块来捕获任何潜在的错误。此外，您可能希望实施各种检查以确保您获得正确类型的数据和您期望需要的所有字段。对于 JSON 正文，您可以创建一个 BaseModel 并使用 Pydantic 的 parse_obj 函数来验证接收到的字典（类似于 this answer 的方法 3）。

关于表单/文件数据，使用request.form() 解析正文将得到一个FormData 对象，这是一个包含文件上传和文本输入的不可变多字典（即ImmutableMultiDict）。当您发送一些form 输入的值列表或files 列表时，您可以使用multidict 的getlist() 方法来检索该列表。对于文件，这将返回UploadFile 对象的列表，您可以使用与this answer 和this answer 相同的方式遍历文件并检索其内容。除了使用request.form()，您还可以直接从stream 读取请求正文，如this answer 中所述。

应用程序.py

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

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

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

选项 3

我还建议您查看this answer，它提供了有关如何将 JSON 正文和 Form-data/Files 一起发送的解决方案，这可能会让您对要解决的问题有不同的看法（例如，将各种端点的参数声明为 Optional 并检查哪些已收到，哪些未收到——以及使用 Pydantic 的 parse_raw() 方法解析在 Form 参数中传递的 JSON 字符串——可能是解决问题的另一种方法）。

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

测试.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)