如何获取tqdm.wrapattr上传进度并传递给客户端实时展示

解决方案 #

你可以通过自定义文件读取包装层注入进度回调，不需要修改现有GCS上传逻辑，也无需引入可恢复上传的复杂实现，改动量极小：

核心思路 #

requests上传时会持续调用传入的文件对象的read方法读取分片，我们只要在文件对象的read逻辑中加入进度统计和回调触发，就能在每次读取分片时拿到实时进度，再通过你服务端的推送通道（Websocket/SSE等）发给客户端即可。

实现代码 #

1. 定义带回调的文件包装类

class ProgressFileReader:
    def __init__(self, file_obj, total_bytes, callback=None):
        self.file_obj = file_obj
        self.total_bytes = total_bytes
        self.read_bytes = 0
        # 回调函数入参为（已读取字节数，总字节数）
        self.callback = callback

    def read(self, size=-1):
        chunk = self.file_obj.read(size)
        if chunk:
            self.read_bytes += len(chunk)
            if self.callback:
                self.callback(self.read_bytes, self.total_bytes)
        return chunk

    # 代理原文件对象的上下文管理和其他属性
    def __enter__(self):
        self.file_obj.__enter__()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        return self.file_obj.__exit__(exc_type, exc_val, exc_tb)

    def __getattr__(self, attr):
        return getattr(self.file_obj, attr)

2. 定义进度推送回调

替换下方示例中的逻辑为你自己的服务端推送逻辑即可：

def push_to_client(read_bytes, total_bytes):
    progress = round(read_bytes / total_bytes * 100, 2)
    # 示例：通过websocket推送给指定客户端
    # client_ws.send(json.dumps({"progress": progress, "file_id": your_file_id}))

3. 改造现有上传逻辑

仅需在原文件对象外层套一层我们自定义的包装类即可，可同时保留tqdm的终端进度展示：

object_address = str(uuid.uuid4())
upload_url, upload_method = get_upload_url(object_address) # fetches signed upload URL 
size = os.path.getsize(filename)
with open(filename, "rb") as in_file:
    total_bytes = os.fstat(in_file.fileno()).st_size
    # 注入自定义进度包装
    progress_file = ProgressFileReader(in_file, total_bytes, callback=push_to_client)
    with tqdm.wrapattr(
        progress_file,
        "read",
        total=total_bytes,
        miniters=1,
        desc="Uploading to my bucket",
    ) as file_obj:
        response = requests.request(
            method=upload_method,
            url=upload_url,
            data=file_obj,
            headers={"Content-Type": "application/octet-stream"},
        )
        response.raise_for_status()

return object_address, size

优化建议 #

• 可在回调中加频率控制，比如进度变化超过1%再推送，避免产生过多推送消息占用带宽
• 如果不需要终端的tqdm进度展示，可直接把progress_file传给requests.request的data参数，无需再套tqdm的包装
• 分片大小由requests内部默认逻辑控制，不需要额外调整，如果你需要自定义分片大小，可在requests.request中添加chunk_size参数

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