如何用aiohttp发送multipart/form-data？请求400错误排查

需要构造并发送如下格式的multipart/form-data请求体：

--XXX
Content-Disposition: form-data; name="name"

John
--XXX
Content-Disposition: form-data; name="age"

12
--XXX--

使用aiohttp编写了代码，但服务器返回400错误：
{'statusCode': 400, 'message': ['queueType must be a valid enum value', 'query must be a string'], 'error': 'Bad Request'}

代码如下：

async def run(self, query: str, queue_type: str = "generate", preset: str = '1', style: str = "") -> Run:
    url = "https://fusionbrain.ai/api/v1/text2image/run"
    boundary = uuid.uuid4().hex

    writer = aiohttp.MultipartWriter('form-data', boundary=boundary)
    writer.append(io.BytesIO(queue_type.encode("UTF-8")), {"Content-Disposition": 'form-data; name="queueType"'})
    writer.append(io.BytesIO(query.encode("UTF-8")), {"Content-Disposition": 'form-data; name="query"'})
    writer.append(io.BytesIO(preset.encode("UTF-8")), {"Content-Disposition": 'form-data; name="preset"'})
    writer.append(io.BytesIO(style.encode("UTF-8")), {"Content-Disposition": 'form-data; name="style"'})
    extra_headers = {'Content-Type': f"multipart/form-data; boundary={boundary}"}

    async with self._session.post(url, headers=self._headers | extra_headers, data=writer) as response:
        return Run.from_response(await response.json())

你的代码存在两个关键问题：

1. 手动设置Content-Type头会引发冲突：MultipartWriter会自动生成包含boundary的Content-Type头，手动添加会导致头信息重复或格式错误，服务器无法正确解析表单。
2. 普通字段用append方法不合适：append主要用于处理文件类二进制数据，普通字符串表单字段应该用append_form方法，它会自动处理编码和字段格式，避免手动封装BytesIO带来的潜在问题。

修正后的代码：

async def run(self, query: str, queue_type: str = "generate", preset: str = '1', style: str = "") -> Run:
    url = "https://fusionbrain.ai/api/v1/text2image/run"
    boundary = uuid.uuid4().hex

    writer = aiohttp.MultipartWriter('form-data', boundary=boundary)
    # 用append_form批量添加普通表单字段
    writer.append_form({
        "queueType": queue_type,
        "query": query,
        "preset": preset,
        "style": style
    })
    # 移除手动添加的Content-Type头，由MultipartWriter自动处理
    async with self._session.post(url, headers=self._headers, data=writer) as response:
        return Run.from_response(await response.json())

如果仍出现queueType枚举错误，排查以下几点：

• 确认queue_type的取值是否在服务器允许的枚举范围内，比如是否为"generate"或其他指定值
• 检查字段名是否完全匹配服务器要求（注意大小写：queueType而非queue_type）
• 可以用抓包工具对比正确请求的格式，确保字段名和值的格式一致

内容的提问来源于stack exchange，提问作者Ilya Nikolaev