如何为同时携带表单数据与文件的aiohttp POST请求添加上传进度监控

实现方案 #

核心思路是封装aiohttp原生FormData的异步迭代过程，在每次传输数据块时统计已发送字节量，触发自定义进度回调，完全兼容原有业务逻辑，不需要修改FormData的字段定义。

第一步：实现进度监控包装工具 #

import aiohttp
import json
from typing import Callable

class AsyncUploadProgressWrapper:
    def __init__(self, form_data: aiohttp.FormData, progress_callback: Callable[[int, int], None]):
        self.form_data = form_data
        self.progress_callback = progress_callback
        # 提前计算总上传大小
        self.total_size = len(form_data)
        self.bytes_read = 0

    @property
    def content_type(self):
        # 直接获取原生FormData的Content-Type头，包含正确的boundary参数
        return self.form_data.content_type

    def __aiter__(self):
        self._iter = self.form_data.__aiter__()
        return self

    async def __anext__(self):
        try:
            chunk = await self._iter.__anext__()
            self.bytes_read += len(chunk)
            # 触发进度回调
            self.progress_callback(self.bytes_read, self.total_size)
            return chunk
        except StopAsyncIteration:
            raise

第二步：替换原有上传逻辑 #

仅需要修改少量代码，原有字段定义完全保留：

# 原有FormData构建逻辑不用修改
data = aiohttp.FormData()
data.add_field('title', title)
data.add_field('author', user)
data.add_field('upload_file', open(path, 'rb'))

# 定义进度回调，和之前requests版本的逻辑完全对齐
def upload_progress(bytes_read: int, total_size: int):
    print(f"{total_size} - {bytes_read / total_size:.2%}")

# 包装FormData实现进度监控
wrapped_data = AsyncUploadProgressWrapper(data, upload_progress)

# 发送请求，注意手动指定Content-Type头
async with aiohttp.ClientSession() as up_session:
    async with up_session.post(
        url="http://example.com/upload.php",
        data=wrapped_data,
        headers={"Content-Type": wrapped_data.content_type}
    ) as response:
        resp = await response.text()

resp = json.loads(resp)

补充说明 #

• 如果上传的文件体积较大，建议用aiofiles异步打开文件代替普通open，避免阻塞事件循环，仅需要把open(path, 'rb')替换为aiofiles.open(path, 'rb')即可，其他逻辑不变。
• 如果FormData中包含未提前确定长度的流式内容，len(form_data)会抛出TypeError，这种情况提前手动计算总大小传入包装类即可。

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