422 Unprocessible Entity error when sending List of Form data through Swagger UI docs using FastAPI

Question

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. However, 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 error, as shown below:

Code    Details
422 
Error: Unprocessable Entity

Response body
Download
{
  "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:

Answer 1

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 it for more information and solutions.

In short, if you instead used some other client, such as Python requests, to test the API endpoint, it should work just fine. 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.

Working Example 1 - Using List of Form data

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())

You could also use query parameters instead, which would work just fine with Swagger UI autodocs, if that suits your application. See this answer, as well as this answer, this answer and this answer for more details and examples. Example is given below.

Working Example 2 - Using List of Query parameters

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())