我想在FastAPI中创建一个端点，该端点可能接收(多部分(Form数据或JSON主体。有没有一种方法可以让这样的端点接受，或者检测正在接收哪种类型的数据？
选项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(，包含文件上传和文本输入。当为某些form输入发送值的list或files的列表时，可以使用多维数据集的getlist()方法来检索list。在文件的情况下，这将返回UploadFile对象的list，您可以使用与此答案和此答案相同的方式来循环浏览文件并检索其内容。不使用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请求，另一个处理File/Form数据请求。使用中间件，您可以检查传入请求是否指向您希望用户发送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
备选方案4
我还建议您看看这个答案，它提供了如何在同一请求中同时发送JSON主体和文件/表单数据的解决方案，这可能会让您对试图解决的问题有不同的看法。例如，将各种端点的参数声明为Optional，并检查哪些参数已经接收到，哪些没有来自客户端的请求，以及使用Pydantic的parse_raw()方法来解析Form参数中传递的JSON字符串，这可能是解决问题的另一种方法。
测试选项1、2和；3使用Python请求
测试.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)