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

Question

我想在FastAPI中创建一个端点，该端点可能会接收 multipart/form-data 或JSON主体。有什么方法可以使这样的端点接受或检测正在接收哪种类型的数据？

Answer 1

选项1

您可以拥有a 依赖项函数，您将在其中检查 content-type 请求标头和使用starlette的方法来解析身体 。请注意，仅仅因为请求的 content-type 标题说，例如， application/json ， application/x-www-form-urlencoded 或 muttipart/form-data ，并不总是意味着这是真的，或传入数据是有效的JSON，或者是文件和/或form-data。因此，您应该使用 try-except 块在解析身体时捕获任何潜在错误。另外，您可能需要实施各种检查，以确保您获得正确的数据类型和所需的所有字段。对于JSON主体，您可以创建一个 basemodel 并使用Pydantic的 parse_obj 功能验证接收的字典（类似于此答案 a>）。

关于file/form-data，您可以使用starlette的对象直接直接，更具体地说，request.form() method to parse the body, which will return a

工作示例

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

另一个选项将是具有单个端点，并将您的文件定义为可选（请查看 this答案和 content> content-type application/x-www-form-urlencoded 或 muttipart/form-data （请注意，如果您期望接收任意文件或form-data，则宁愿使用上面的选项1 ）。否则，如果每个定义的参数仍然无（这意味着客户端未在请求正文中包含其中任何一个），那么这可能是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请求，另一个用于处理文件/表单数据请求。使用 middledware ，您可以检查输入请求是否指向您希望用户的路线要发送JSON或文件/表单数据（在下面的示例/路由）中，如果是的，请检查 content-type 与上一个选项类似并重新路由请求/submitjson 或/subtsform endpoint，相应地（您可以通过修改 path> path 属性 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

选项4

我还建议您查看这个答案，该答案提供了有关如何发送JSON身体的解决方案以及文件/form-data在相同的请求中一起，这可能会使您对要解决的问题有不同的看法。例如，将各种端点的参数声明为可选并检查已接收到哪些参数，哪些没有从客户端的请求中 - 以及使用Pydantic的 model_validate_json（）方法要解析以 form 参数传递的JSON字符串，可能是解决问题的另一种方法。有关更多详细信息和示例，请参见上面的链接答案。

测试选项1、2＆amp; 3使用python请求

test.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)