如何使用Python OpenStack SDK实现文件上传进度条展示功能

你可以通过包装原生文件对象的方式自定义读取回调逻辑，完全满足上传进度统计的需求，无需修改Python内置文件对象或OpenStack SDK的源码。

核心实现思路 #

OpenStack SDK上传文件时会调用传入文件对象的read()方法按块读取内容，我们只需要代理原生文件对象，重写read()方法，在每次读取操作完成后统计已读字节数、触发进度更新回调即可。

步骤1：实现可统计读取进度的文件包装类 #

import os

class ProgressFileWrapper:
    def __init__(self, file_path, read_callback):
        self.file = open(file_path, 'rb')
        self.total_size = os.path.getsize(file_path)
        self.read_bytes = 0
        self.read_callback = read_callback
        # 首次回调初始化进度
        self.read_callback(self.read_bytes, self.total_size)

    def read(self, size=-1):
        data = self.file.read(size)
        self.read_bytes += len(data)
        self.read_callback(self.read_bytes, self.total_size)
        return data
    
    # 实现上下文管理器支持with语句
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.file.close()
        return False

    # 代理其他原生文件对象的方法，避免SDK调用其他方法报错
    def __getattr__(self, attr):
        return getattr(self.file, attr)

步骤2：结合已有的进度条代码使用 #

以下是完整的使用示例：

# 原有进度条、颜色代码可直接复用
ESCAPE = {
    'c': { # colours
        'fg': { # foreground
            'b': '30', # black
            'r': '31', # red
            'g': '32', # green
            'y': '33', # yellow
            'u': '34', # blue
            'p': '35', # purple
            'c': '36', # cyan
            'w': '37', # white
        },
        'bg': { # background
            'b': '40', # black
            'r': '41', # red
            'g': '42', # green
            'y': '43', # yellow
            'u': '44', # blue
            'p': '45', # purple
            'c': '46', # cyan
            'w': '47', # white
        }
    },
    's': { # style
        'n': '0',  # none
        'b': '1',  # bold
        'u': '2',  # underline
        'n1': '3', # negative1
        'n2': '5', # negative2
    },
    't': '\033[{s};{fg};{bg}m' # template 
}

def colour(fg='w', bg='b', s='n'):
    return ESCAPE['t'].format(
            s=ESCAPE['s'][s],
            fg=ESCAPE['c']['fg'][fg],
            bg=ESCAPE['c']['bg'][bg]
    )

def get_progress_bar(percent, max_col=78):
    filled_len = int(max_col * percent / 100)
    bar = ' ' * filled_len
    togo = ' ' * (max_col - filled_len)
    return f"[{colour(bg='g')}{bar}{colour(bg='r')}{togo}{colour()}|{percent:6.2f}%]"

# 上传调用示例
if __name__ == "__main__":
    # 要上传的镜像路径
    image_path = "/path/to/your/image.iso"
    
    def update_progress(read_bytes, total_size):
        percent = (read_bytes / total_size) * 100
        # 清除当前行输出新进度
        print('\r' + get_progress_bar(percent), end='', flush=True)
        # 读完后换行
        if read_bytes >= total_size:
            print()
    
    # 用包装类代替原生open返回的文件对象
    with ProgressFileWrapper(image_path, update_progress) as f:
        # 这里直接把f传给OpenStack SDK的上传方法即可，比如：
        # conn.image.create_image(name="test-image", data=f, disk_format="iso")
        # 模拟读取操作，SDK内部会自动调用read触发进度更新
        while f.read(1024*1024):
            pass

注意事项 #

• 如果上传时SDK会对文件内容做哈希校验等额外读取操作，你可以在包装类中加开关控制是否统计本次读取的字节数，避免进度重复计算
• 对于超大文件，你可以降低进度条的刷新频率，比如每读取1MB才更新一次UI，避免频繁打印导致性能下降

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