422 Unprocessible Entity error when sending List of Form data through Swagger UI docs using FastAPI
I created an API, using FastAPI, for transcoding video files like adaptive bitrate of youtube. Currently, I am trying to set resolution values for each file. But, whenever I am passing multiple resolutions as a List of Form data through Swagger UI autodocs, I am recieving a 422 unprocessible entity error.
Here is my code:
async def transcode_video(input_path, output_folder, res, unique_id, total_files, pbar):
# Use asyncio for command execution with progress bar
output_path = os.path.join(output_folder, f"{res}p.m3u8")
# Calculate the target size
target_size = calculate_target_size(input_path)
cmd_duration = f"ffprobe -v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 {input_path}"
total_duration = float(subprocess.check_output(cmd_duration, shell=True, text=True).strip())
cmd = (
f"ffmpeg -i {input_path} -vf scale=-2:{res} -c:a aac -b:a 128k "
f"-g 50 -hls_time 1 -hls_list_size 0 "
f"-crf 23 -b:v 100k -fs {target_size} "
f"-hls_segment_filename \"{output_path.replace('.m3u8', '_%03d.ts')}\" "
f"{output_path}"
)
process = await asyncio.create_subprocess_shell(
cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
while True:
line = await process.stderr.readline()
if not line:
break
line = line.decode().strip()
if "time=" in line:
# Extracting the time progress from FFmpeg output
time_str = line.split("time=")[1].split()[0]
current_time = sum(x * float(t) for x, t in zip([3600, 60, 1], time_str.split(":")))
progress = (current_time / total_duration) * 100
pbar.update(progress - pbar.n)
# Wait for the transcoding process to complete
await process.wait()
if process.returncode != 0:
raise HTTPException(status_code=500, detail="Video transcoding failed.")
pbar.close()
# Increment the total number of transcoded files
total_files[0] += 1
@app.post("/transcode/")
async def transcode_video_endpoint(files: List[UploadFile] = File(...), resolutions: List[int] = None):
# Counters for transcoded videos
total_files = [0]
# Iterate over each file
for file in files:
# Check if the file is a valid video file based on its extension
valid_video_extensions = {".mp4", ".avi", ".mkv", ".mov", ".wmv", ".flv"}
if not any(file.filename.lower().endswith(ext) for ext in valid_video_extensions):
print(f"Skipping non-video file: {file.filename}")
continue
# Assign a unique ID for each file
unique_id = str(uuid.uuid4())
# Log the filename and unique ID
print(f"Processing file: {file.filename} with unique ID: {unique_id}")
# Create a folder for the unique ID
unique_id_folder = os.path.join(OUTPUT_FOLDER, unique_id)
Path(unique_id_folder).mkdir(parents=True, exist_ok=True)
# Save the uploaded file
input_path = os.path.join(UPLOAD_FOLDER, file.filename)
with open(input_path, "wb") as video_file:
video_file.write(file.file.read())
# Check if the file is a valid video file using ffprobe
try:
subprocess.run(
["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_type", "-of", "csv=p=0", input_path],
check=True, capture_output=True
)
except subprocess.CalledProcessError:
print(f"Skipping non-video file: {file.filename}")
continue
# Determine the resolutions to transcode based on the provided or default resolution
resolutions_to_transcode = [res for res in [240, 360, 480, 720] if resolutions is None or res in resolutions]
# If resolutions is not exactly 360, 480, 720, or 1080, transcode to the nearest lower resolution
if resolutions is not None:
resolutions_to_transcode = [get_closest_lower_resolution(res) for res in resolutions]
# Transcode the video into the specified resolutions
for res in resolutions_to_transcode:
output_folder = os.path.join(unique_id_folder, f"{res}p")
Path(output_folder).mkdir(parents=True, exist_ok=True)
# Call the transcode_video function with tqdm progress bar
with tqdm(total=100, desc=f"Transcoding {res}p", position=0, leave=True) as pbar:
await transcode_video(input_path, output_folder, res, unique_id, total_files, pbar)
# Create index.m3u8 file after transcoding all resolutions
create_index_m3u8(unique_id, resolutions_to_transcode)
return JSONResponse(content={"message": f"{total_files[0]} videos transcoded."})
If I provide only one resolution value, it works just fine. But, if I provide a list of resolutions for all the uploaded videos, I am getting 422 Unprocessable Entity error, as shown below:
{
"detail": [
{
"loc": [
"body",
"resolutions",
0
],
"msg": "value is not a valid integer",
"type": "type_error.integer"
}
]
}
I am using python 3.8.18. Can you help me please? I tried with list, object but always getting the error. My input format is shown below:
This is a known issue in Swagger UI and can't really tell why its hasn't been fixed yet. When sending a list of data (reagrdless of type; it can be either int or str), they are sent as a single string value instead of separate values. It has been exaplained in more detail in this answer; hence, please have a look at that answer for more details and solutions.
In short, if you instead used some other client/library to send HTTP requests, such as Python requests or httpx, it should work just fine.
Working Example 1 - Using List of Form data
The example below is based on Method 1 of this answer—please have a look at that answer, as you might find helpful details and examples.
app.py
from fastapi import FastAPI, Form, File, UploadFile
from typing import List
app = FastAPI()
@app.post("/submit")
def submit(
files: List[UploadFile] = File(...),
resolutions: List[int] = Form(...)
):
return {
"Filenames": [file.filename for file in files],
"resolutions": resolutions
}
test.py
import requests
url = 'http://127.0.0.1:8000/submit'
files = [('files', open('a.txt', 'rb'))]
data = {"resolutions": [720, 480]}
r = requests.post(url=url, data=data, files=files)
print(r.json())
Working Example 2 - Using List of Query parameters
You could also use query parameters instead, which would work just fine with Swagger UI autodocs, if that suits your application. See this answer and this answer, as well as this and this for more details and examples.
app.py
from fastapi import FastAPI, File, UploadFile, Query
from typing import List
app = FastAPI()
@app.post("/submit")
def submit(
files: List[UploadFile] = File(...),
resolutions: List[int] = Query(...)
):
return {
"Filenames": [file.filename for file in files],
"resolutions": resolutions
}
test.py
import requests
url = 'http://127.0.0.1:8000/submit'
files = [('files', open('a.txt', 'rb'))]
data = {"resolutions": [720, 480]}
r = requests.post(url=url, params=data, files=files)
print(r.json())