集成 API 调用的进度条

Question

背景： 我已经看到很多将进度条集成到 for 循环中的示例，但是对于我的用例没有任何帮助，因此我正在寻找一些建议。

对于我的用例，我正在调用 API 并测试 meta 是否在响应中（meta = 我需要的数据）。如果 meta 不在 API 响应中，那么 API return 是一个名为 percent_complete 的密钥对值，它表示我正在尝试 return仍在聚合中，并提供数据聚合进度的值。

当前代码：

def api_call():

    key, secret, url = ini_reader()
    endpoint_url = endpoint_initializer()
    
    while True:
        response = requests.get(url = endpoint_url, auth = HTTPBasicAuth(key, secret), headers = {"vendor-firm": "111"})
        api_response = json.loads(response.text)

        if "meta" not in api_response:
            id_value = "id"
            res1 = [val[id_value] for key, val in api_response.items() if id_value in val]
            id_value = "".join(res1)
            percent_value = "percent_complete"
            res2 = api_response["data"]["attributes"].get("percent_complete", '')*100
            print(f' Your data request for: {id_value} is {res2}% complete!')
            time.sleep(60)

       elif "meta" in api_response:
            
     return api_response

我想要实现的目标： {res2} *100 给出了百分比，我想在进度条中使用该百分比来衡量进度。

任何人都可以建议使用合适的依赖项吗？

Answer 1

您可以使用 Enlighten 库。您可以保留您的打印语句并同时拥有多个进度条，而无需进行任何其他更改。下面是您如何实施它的示例。

根据你的示例，它看起来 id_value 发生了变化，所以我写了这样的示例。如果它没有改变，你可以在描述中使用它。如果你有多个，你可能想为每个创建一个进度条。如果您想在完成后删除进度条，只需将 leave=False 添加到 manager.Counter()。

库非常可定制，documentation has a lot of examples。

import enlighten

BAR_FORMAT = u'{id_value} {percentage:3.0f}%|{bar}| ' u'[{elapsed}<{eta}, {rate:.2f} %/s]'

manager = enlighten.get_manager()

def api_call():
    pbar = manager.counter(total=100, bar_format=BAR_FORMAT)
    ...
    while True:
        ...
    if "meta" not in api_response:
        ...
        pbar.count = res2
        pbar.update(incr=0, id_value=id_value)

    else:
        ...
        pbar.count = 100
        pbar.update(incr=0, id_value=id_value)

    pbar.close()
    return api_response

Answer 2

感谢 Aviso，为了大家的利益，这里是完成的功能 -

def api_call():
    endpoint_url = endpoint_initializer()
    key, secret, url = ini_reader()
    
    BAR_FORMAT = u'{id_value} {percentage:3.0f}%|{bar}| ' u'[{elapsed}<{eta}, {rate:.2f} %/s]'
    manager = enlighten.get_manager()
    
    date = dt.datetime.today().strftime("%Y-%m-%d")
    print("------------------------------------\n","API URL constructed for:", date, "\n------------------------------------")
    print("-------------------------------------------------------------\n","Endpoint:", endpoint_url, "\n-------------------------------------------------------------") 
    
    pbar = manager.counter(total=100, bar_format=BAR_FORMAT)

    while True:
        response = requests.get(url = endpoint_url, auth = HTTPBasicAuth(key, secret), headers = {"vendor-firm": "381"})
        api_response = json.loads(response.text) 
        
        if "meta" not in api_response:
            id_value = "id"
            res1 = [val[id_value] for key, val in api_response.items() if id_value in val]
            id_value = "".join(res1)
            percent_value = "percent_complete"
            res2 = api_response["data"]["attributes"].get("percent_complete", '')*100
            pbar.count = res2
            pbar.update(incr=0, id_value=id_value)
            time.sleep(60)
        
        elif "meta" in api_response:
            pbar.count = 100
            pbar.update(incr=0, id_value=id_value)
            pbar.close()
            return api_response

集成 API 调用的进度条

Integrating Progress Bar for API Call

python

pandas

progress-bar