I would like to create an endpoint in FastAPI that might receive either multipart/form-data or JSON body. Is there a way I can make such an endpoint accept either, or detect which type of data is receiving?
You could have a dependency function, where you would check the value of the Content-Type request header and parse the body using Starlette's methods, accordingly. Note that just because a request's Content-Type header says, for instance, application/json, application/x-www-form-urlencoded or multipart/form-data, doesn't always mean that this is true, or that the incoming data is a valid JSON, or File(s) and/or form-data. Hence, you should use a try-except block to catch any potential errors when parsing the body. Also, you may want to implement various checks to ensure that you get the correct type of data and all the fields that you expect to be required. For JSON body, you could create a BaseModel and use Pydantic's parse_obj function to validate the received dictionary (similar to Method 3 of this answer).
Regarding File/Form-data, you can use Starlette's Request object directly, and more specifically, the request.form() method to parse the body, which will return a FormData object that is an immutable multidict (i.e., ImmutableMultiDict) containing both file uploads and text input. When you send a list of values for some form input, or a list of files, you can use the multidict's getlist() method to retrieve the list. In the case of files, this would return a list of UploadFile objects, which you can use in the same way as this answer and this answer to loop through the files and retrieve their content. Instead of using request.form(), you could also read the request body directly from the stream and parse it using the streaming-form-data library, as demonstrated in this answer.
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
Another option would be to have a single endpoint, and have your File(s) and/or Form data parameters defined as Optional (have a look at this answer and this answer for all the available ways on how to do that). Once a client's request enters the endpoint, you could check whether the defined parameters have any values passed to them, meaning that they were included in the request body by the client and this was a request having as Content-Type either application/x-www-form-urlencoded or multipart/form-data (Note that if you expected to receive arbitrary file(s) or form-data, you should rather use Option 1 above ). Otherwise, if every defined parameter was still None (meaning that the client did not include any of them in the request body), then this was likely a JSON request, and hence, proceed with confirming that by attempting to parse the request body as 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')
Another option would be to define two separate endpoints; one to handle JSON requests and the other for handling File/Form-data requests. Using a middleware, you could check whether the incoming request is pointing to the route you wish users to send either JSON or File/Form data (in the example below that is / route), and if so, check the Content-Type similar to the previous option and reroute the request to either /submitJSON or /submitForm endpoint, accordingly (you could do that by modifying the path property in request.scope, as demonstrated in this answer). The advantage of this approach is that it allows you to define your endpoints as usual, without worrying about handling errors if required fields were missing from the request, or the received data were not in the expected format.
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
I would also suggest you have a look at this answer, which provides solutions on how to send both JSON body and Files/Form-data together in the same request, which might give you a different perspective on the problem you are trying to solve. For instance, declaring the various endpoint's parameters as Optional and checking which ones have been received and which haven't from a client's request—as well as using Pydantic's model_validate_json() method to parse a JSON string passed in a Form parameter—might be another approach to solving the problem. Please see the linked answer above for more details and examples.
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)