python openstack SDK通过进度条显示文件上传答案

【问题标题】：python openstack SDK show file upload via progress barpython openstack SDK通过进度条显示文件上传
【发布时间】：2021-11-14 10:00:07
【问题描述】：

使用 python openstack SDK，您可以逐块下载图像，这使我可以显示进度条。

上传文件时您似乎不能这样做。

我想要的是能够显示正在上传的文件的进度条（以与正在下载的文件相同的方式）。

有没有办法将一种“文件读取”回调附加到文件对象？所以每次读取时我都可以显示进度条？

我确实查看了 python 文件对象文档，但我真的找不到任何确定的东西。

附言

有兴趣的可以看我的代码：

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 spinner(msg: str):
    template = f"{colour(fg='p', s='b')}{'{}'}{colour()} {msg}"
    while True:
        for spin in '⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏':
            yield template.format(spin)

def progress(end: int, max_col=78):
    template = f"[{colour(bg='g')}{'{}'}{colour(bg='r')}{'{}'}{colour()}|{'{}'}]"
    for p in range(1, end):
        bar = ' ' * int(max_col * p/(end-1))
        togo = ' ' * int(max_col - len(bar))
        perc = "%6.2f %%" % (p/(end-1)*100)
        yield template.format(bar, togo, perc)

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]
    )

【问题讨论】：

标签： python file stream openstacksdk

【解决方案1】：

因此，经过大量搜索后，我找到了一种方法来满足我的要求（here & here）。

代码：

#!/usr/bin/env python3
from types import MethodType

def hook(hookfunk, oldfunc):
    def merged(self, *args, **kwargs):
        hookfunk(self, *args, **kwargs)
        return oldfunc(*args, **kwargs)
    return  MethodType(merged, oldfunc.__self__)


f = open('cirros.qcow2', 'rb')

def new_read(self, size):
    print(self)
    # Or do other code, like updating a progress bar

f.read = hook(new_read, f.read)

f.read()

【讨论】：