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

问题描述：

我想在 FastAPI 中创建一个可以接收multipart/form-dataJSON 主体的端点。有没有办法让这样的端点接受任一类型，或者检测正在接收哪种类型的数据？

解决方案 1：

选项 1

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

关于文件/表单数据，您可以直接使用Starlette的Request对象，更具体地说，request.form()使用方法来解析主体，它将返回一个FormData不可变的多字典对象（即ImmutableMultiDict），包含文件上传和文本输入。当您发送一些list输入的值form或列表时files，您可以使用多字典的getlist()方法来检索list。对于文件，这将返回一个list对象UploadFile，您可以按照与此答案和此答案相同的方式使用这些对象来循环遍历文件并检索其内容。除了使用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中所示）。这种方法的优点是，它允许您像往常一样定义端点，而不必担心如果请求中缺少必填字段或收到的数据不是预期的格式时处理错误。

工作示例

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()方法来解析传入参数的 JSON 字符串Form——可能是解决问题的另一种方法。有关更多详细信息和示例，请参阅上面链接的答案。

使用 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)