Python异步请求怎么发送multipart/form-data

在 Python 异步编程中发送 multipart/form-data 表单，最直接的方法是使用当前主流的异步 HTTP 库（如 HTTPX 或 aiohttp），通过其内置参数或表单构建对象将文件和字段传递给请求方法。只要交由底层客户端自动计算 Boundary 分隔符与请求头，配合异步上下文管理器妥善管理连接与文件句柄，就能安全高效地完成文本与二进制文件的异步上传。

一、理解 multipart/form-data 的报文结构与工作逻辑

向服务器提交包含文件或混合格式数据的表单时，标准的 URL 编码表单（application/x-www-form-urlencoded）无法承载大量的二进制内容。HTTP 协议规范为此设计了 multipart/form-data 媒体类型，其核心是将请求体拆分为多个独立的块（Parts），各块之间通过一段全局唯一的随机字符串（Boundary）进行隔离。

在异步请求生命周期中，客户端不会一次性把整个请求体阻塞式地写入套接字，而是借助异步 I/O 在数据段就绪时分段写入。理解客户端在后台如何拼接这些数据段，有助于在遇到参数解析错误或后端校验异常时快速定位根因。

二、使用 HTTPX 发送异步 multipart 请求

HTTPX 是目前 Python 社区中现代化程度高、API 与 requests 保持高度一致的异步 HTTP 库。如果此前熟悉 requests 的语法，使用 HTTPX 进行异步文件上传几乎没有额外记忆成本。

1. 安装依赖

pip install httpx

2. 核心操作步骤

在异步上下文中，使用 HTTPX 发送 multipart 数据通常分为四个阶段：准备数据元组、建立异步客户端会话、执行异步非阻塞发送、自动释放上下文资源。

3. 单文件与普通字段混传示例

当表单中既有普通字段（如用户标识、分类标签）又有上传文件时，使用 data 参数传递普通字典，使用 files 参数传递文件元组：

import asyncio
import httpx

async def upload_single_file():
    target_url = "https://httpbin.org/post"
    
    # 模拟从磁盘或内存中获取的二进制字节流
    file_bytes = b"%PDF-1.4 sample binary data..."
    
    # 构造 files 字典：字段名 -> (文件名, 文件内容字节, MIME 类型)
    files = {
        "file": ("report.pdf", file_bytes, "application/pdf")
    }
    
    # 构造普通表单字段
    data = {
        "user_id": "10086",
        "category": "finance"
    }

    async with httpx.AsyncClient() as client:
        # 发送异步 POST 请求，不手动指定 headers['Content-Type']
        response = await client.post(target_url, data=data, files=files)
        
        # 确认响应状态码
        if response.status_code == 200:
            result = response.json()
            print("上传成功，服务器收到的表单字段:", result.get("form"))
            print("上传成功，服务器收到的文件列表:", result.get("files"))
        else:
            print(f"上传失败，状态码: {response.status_code}")

if __name__ == "__main__":
    asyncio.run(upload_single_file())

4. 批量多文件上传

如果同一个表单字段需要上传多个文件，或者多个不同字段对应不同文件，可以向 files 传入列表或在同一个字段名下挂载多项：

async def upload_multiple_files():
    target_url = "https://httpbin.org/post"
    
    # 使用包含元组的列表表示多个同名字段文件
    files = [
        ("documents", ("page1.png", b"fake-png-binary-1", "image/png")),
        ("documents", ("page2.png", b"fake-png-binary-2", "image/png")),
        ("attachment", ("notes.txt", b"meeting notes content", "text/plain"))
    ]
    
    data = {"batch_id": "batch_20261026"}

    async with httpx.AsyncClient() as client:
        response = await client.post(target_url, data=data, files=files)
        response.raise_for_status()
        print("批量上传完成，响应大小:", len(response.content))

三、使用 aiohttp 处理 multipart/form-data

aiohttp 是 Python 异步网络框架中历史悠久且高性能的选择。与 HTTPX 传参即封装的风格不同，aiohttp 推荐显式使用 aiohttp.FormData 对象构建复杂表单。

1. 安装依赖

pip install aiohttp

2. 使用 FormData 对象构建与发送

通过 FormData.add_field() 方法，可以细粒度地指定每个字段的键名、数据内容、文件名、MIME 类型乃至内容传输编码：

import asyncio
import aiohttp

async def upload_with_aiohttp():
    target_url = "https://httpbin.org/post"

    # 1. 创建 FormData 容器
    form = aiohttp.FormData()

    # 2. 追加普通文本字段
    form.add_field("user_id", "4096")
    form.add_field("description", "用户头像更新")

    # 3. 追加文件字段（传入字节流、自定义文件名和 content_type）
    avatar_bytes = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR..."
    form.add_field(
        name="avatar",
        value=avatar_bytes,
        filename="avatar_thumb.png",
        content_type="image/png"
    )

    # 4. 在异步客户端会话中发起请求
    async with aiohttp.ClientSession() as session:
        # data 直接接收 FormData 实例，aiohttp 会自动计算边界并流式发送
        async with session.post(target_url, data=form) as response:
            status = response.status
            response_json = await response.json()
            print(f"aiohttp 请求响应状态: {status}")
            print("服务端解析结果:", response_json.get("files"))

if __name__ == "__main__":
    asyncio.run(upload_with_aiohttp())

3. HTTPX 与 aiohttp 的实现差异对比

在面对不同架构或业务规模时，选择合适的客户端能减少重复封装。下面是两种方案的技术特性对比：

四、大文件上传与异步文件 I/O

当上传的文件体积达到数百兆（甚至数吉字节）时，直接调用 open().read() 将文件全部读取为 bytes 会迅速消耗服务器内存甚至引发 OOM（内存溢出）。此外，标准的内置 open() 是同步磁盘 I/O，在大文件读写期间会阻塞 Python 的主事件循环，导致其他并发协程停滞。

解决大文件上传的正确策略是：流式分块读取与异步流式传输。

1. HTTPX 中的异步流式上传

HTTPX 允许直接将一个异步生成器作为文件内容传入，从而按需分块读取并发送：

import asyncio
import httpx

# 模拟异步生成器分块读取大文件
async def file_chunk_generator(filepath: str, chunk_size: int = 65536):
    # 此处若在生产环境，可配合 aiofiles 库进行完全无阻塞读取
    with open(filepath, "rb") as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

async def stream_upload_large_file(filepath: str):
    target_url = "https://example.com/api/upload-stream"
    
    # 将异步迭代器传给 content 参数或包装在 files 映射中
    files = {
        "large_file": ("video.mp4", file_chunk_generator(filepath), "video/mp4")
    }

    async with httpx.AsyncClient(timeout=300.0) as client:
        # 注意：大文件上传需适当调大 timeout 参数
        response = await client.post(target_url, files=files)
        print("流式上传状态:", response.status_code)

五、常见问题与排查清单

在发送异步 multipart/form-data 时，常见的错误往往并非网络问题，而是由不当的参数传递或头部覆盖导致的。以下是四个最高频的踩坑场景与核对清单：