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

在 FastAPI 中，您可以创建可以接受表单或 JSON 正文的端点使用不同的方法。以下是几个选项：

选项 1：使用依赖函数

此选项涉及创建一个依赖函数，该函数检查 Content-Type 请求标头的值并使用 Starlette 的方法解析正文，相应地。

from fastapi import FastAPI, Depends, Request
from starlette.datastructures import FormData

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':
        return await request.json()
    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')
        return msg

选项 2：定义单独的端点

另一个选项是使用单个端点，并将文件和/或表单数据参数定义为可选。如果任何参数有值传递给它们，则意味着请求是 application/x-www-form-urlencoded 或 multipart/form-data。否则，它可能是一个 JSON 请求。

from fastapi import FastAPI, UploadFile, File, Form
from typing import Optional, List

app = FastAPI()

@app.post('/')
async def submit(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
        data = await request.json()
        return {'JSON': data}

选项 3：使用中间件

您还可以使用中间件检查传入请求并将其重新路由到 /submitJSON 或 /submitForm 端点，具体取决于在请求的 Content-Type 上。

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@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

测试选项

您可以使用Python的requests库测试上述选项：

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)