如何用Python Requests调用接收Multipart表单的FastAPI接口？

问题场景 #

现有如下FastAPI接口：

@app.post("/process-file")
async def process_file(
    filename: str,
    config: Config,
    file: UploadFile = File(...),
) -> StreamingResponse:

使用requests库发送请求时，尝试将filename作为URL参数、config通过json参数传递，代码如下：

config = {
    "threshold": 0.5,
    "black_background": True,
    "bounding_boxes": False,
    "text": True,
    "alpha": 1,
    "beta": 0.6,
    "gamma": 0,
}

for filename in os.listdir(INPUT_DIR):
    path = os.path.join(INPUT_DIR, filename)
    with open(path, "rb") as f:
        print(f"Sending {filename} to the API...")
        params = {"filename": filename}
        files = {"file": f}
        response = requests.post(
            API_URL_FILE,
            params=params,
            json=config,
            files=files,
        )

请求返回422错误：

Error: 422, [{'loc': ['body', 'config'], 'msg': 'field required', 'type': 'value_error.missing'}]

问题原因 #

当请求中包含文件（使用files参数）时，requests会自动将请求编码为multipart/form-data格式，但此时使用json参数会同时设置Content-Type: application/json请求头，导致请求格式冲突。FastAPI无法在multipart请求中解析到JSON格式的config参数，因此返回字段缺失的错误。

解决方案 #

步骤1：修改FastAPI接口 #

将filename和config改为表单字段接收，config以JSON字符串形式传递，在接口内部解析为Config对象：

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import json

app = FastAPI()

class Config(BaseModel):
    threshold: float
    black_background: bool
    bounding_boxes: bool
    text: bool
    alpha: float
    beta: float
    gamma: int

@app.post("/process-file")
async def process_file(
    filename: str = Form(...),
    config: str = Form(...),
    file: UploadFile = File(...),
) -> StreamingResponse:
    # 将JSON字符串解析为Config对象
    config_obj = Config.parse_raw(config)
    # 此处添加你的文件处理逻辑
    # 示例：返回处理后的文件流
    return StreamingResponse(file.file, media_type="application/octet-stream")

步骤2：调整requests请求代码 #

将filename和JSON字符串化的config放到data参数中，同时传递文件对象：

import json
import os
import requests

config = {
    "threshold": 0.5,
    "black_background": True,
    "bounding_boxes": False,
    "text": True,
    "alpha": 1,
    "beta": 0.6,
    "gamma": 0,
}

INPUT_DIR = "你的输入目录路径"
API_URL_FILE = "http://你的API地址/process-file"

for filename in os.listdir(INPUT_DIR):
    path = os.path.join(INPUT_DIR, filename)
    with open(path, "rb") as f:
        print(f"Sending {filename} to the API...")
        # 构造表单数据，将config转为JSON字符串
        data = {
            "filename": filename,
            "config": json.dumps(config)
        }
        # 传递文件，建议指定文件名与接口参数对应
        files = {"file": (filename, f)}
        response = requests.post(
            API_URL_FILE,
            data=data,
            files=files,
        )
        # 处理响应
        if response.ok:
            # 示例：保存响应的流文件
            with open(f"processed_{filename}", "wb") as out_file:
                out_file.write(response.content)
        else:
            print(f"请求失败：{response.status_code}, {response.json()}")

内容的提问来源于stack exchange，提问作者Reuben Dubester