如何追踪使用Python SDK的Presigned POST上传AWS S3的进度？

嘿，我之前也碰到过这个问题！用Presigned POST上传S3的时候，boto3确实没有内置的进度追踪钩子，但我们可以在发送POST请求的环节自己实现，核心思路就是在文件上传过程中统计已发送的字节数，进而计算进度。

Presigned POST本质是构造一个带签名的表单，再通过HTTP POST请求把文件传到S3，所以我们只需要在发送这个请求的步骤里，给文件读取过程加上进度监听逻辑就行。下面是具体的实现步骤：

1. 先生成Presigned POST表单 #

这部分和你之前的代码逻辑一致，用boto3生成带签名的上传表单：

import boto3
from botocore.exceptions import ClientError

s3_client = boto3.client('s3')

def create_presigned_post(bucket_name, object_name, fields=None, conditions=None, expiration=3600):
    try:
        response = s3_client.generate_presigned_post(
            bucket_name,
            object_name,
            Fields=fields,
            Conditions=conditions,
            ExpiresIn=expiration
        )
    except ClientError as e:
        print(f"生成Presigned POST失败: {e}")
        return None
    return response

2. 定义进度追踪的回调函数 #

写一个简单的回调函数，用来接收已上传字节数和总字节数，计算并打印进度：

def upload_progress_callback(uploaded_bytes, total_bytes):
    progress_percent = (uploaded_bytes / total_bytes) * 100
    # 用\r实现进度条刷新效果
    print(f"上传进度: {progress_percent:.2f}% ({uploaded_bytes}/{total_bytes} 字节)", end='\r')

3. 包装文件对象并发送POST请求 #

这里用requests库来发送请求（比自带的urllib更灵活），我们需要自定义一个文件类，在每次读取文件块的时候触发进度回调：

import os
import requests

class ProgressTrackingFile:
    def __init__(self, file_obj, total_size, callback):
        self.file_obj = file_obj
        self.total_size = total_size
        self.callback = callback
        self.uploaded = 0

    def read(self, chunk_size=1024*1024):  # 按1MB块读取
        data = self.file_obj.read(chunk_size)
        if data:
            self.uploaded += len(data)
            self.callback(self.uploaded, self.total_size)
        return data

然后用这个类包装你的文件，再发送Presigned POST请求：

# 生成上传表单
presigned_post = create_presigned_post('你的存储桶名称', '要上传的对象键')
if not presigned_post:
    exit("无法生成Presigned POST表单")

# 准备要上传的文件
file_path = '/path/to/your/file'
total_file_size = os.path.getsize(file_path)

with open(file_path, 'rb') as f:
    # 包装文件对象，绑定进度回调
    tracked_file = ProgressTrackingFile(f, total_file_size, upload_progress_callback)
    # 发送POST请求
    response = requests.post(
        presigned_post['url'],
        data=presigned_post['fields'],
        files={'file': tracked_file}
    )

# 检查上传结果
if response.status_code == 204:
    print("\n上传完成！")
else:
    print(f"\n上传失败，状态码: {response.status_code}，响应内容: {response.text}")

进阶：用tqdm实现美观的进度条 #

如果觉得自己打印进度不够好看，可以用tqdm库快速实现带样式的进度条：

from tqdm import tqdm

# 初始化进度条
progress_bar = tqdm(total=total_file_size, unit='B', unit_scale=True, desc="上传中")

def tqdm_progress_callback(uploaded_bytes, total_bytes):
    # 更新进度条到当前已上传字节数
    progress_bar.update(uploaded_bytes - progress_bar.n)

# 后续步骤和之前一样，只是把回调换成这个tqdm的版本

这个方法完全基于boto3和requests，不需要额外的S3相关依赖，而且能精准追踪每一块文件的上传进度，亲测有效！

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