python使用rich.progress打印彩色进度条

常规的同步方法和for循环的进度，使用tqdm能很直观地展示进度；而一些异步协程或难以预估进度的，可以考虑使用rich.progress

Ref: https://typer.tiangolo.com/tutorial/progressbar/#progress-bar

案例一：左侧展示旋转的小圈圈

from contextlib import contextmanager
from typing import Generator# pip install asynctor httpx rich
import asynctor
import httpx
from rich.progress import Progress, SpinnerColumn@contextmanager
def progressbar(msg: str, color: str = "cyan", transient=False
) -> Generator[None, None, None]:"""Spinner风格的七彩进度条:param msg: 进度条描述文字:param color: 颜色，如：'blue':param transient: 任务完成后是否清除进度条信息"""with Progress(SpinnerColumn(), *Progress.get_default_columns(), transient=transient) as progress:progress.add_task(f"[{color}]{msg}...", total=None)yield@asynctor.timeit
async def fetch(url) -> int:async with httpx.AsyncClient(verify=False, follow_redirects=True) as client:r = await client.get(url)return r.status_codeasync def main():url = "https://pypi.org"with progressbar(f"Fetching {url}"):status_code = await fetch(url)print(f"{status_code = }")url = "https://fastapi.tiangolo.com/"with progressbar(f"Fetching {url}", color="green", transient=True):status_code = await fetch(url)print(f"{url = }; {status_code = }")if __name__ == "__main__":asynctor.run(main)

效果如下：

案例二：展示百分比

#!/usr/bin/env python
from contextlib import asynccontextmanager
from typing import AsyncGenerator# pip install asynctor httpx rich
import anyio
import asynctor
import httpx
from rich.progress import Progress@asynccontextmanager
async def percentbar(msg: str, seconds=5, color: str = "cyan", transient=False
) -> AsyncGenerator[None, None]:"""Spinner风格的七彩进度条:param msg: 进度条描述文字:param seconds: 任务总时长:param color: 颜色，如：'blue':param transient: 任务完成后是否清除进度条信息"""total = seconds * 100async def play(progress, task, expected=1 / 2, thod=0.8):# 改变进度条速率，前面1/2的时间完成80%的进度cost = seconds * expectedquick = int(total * thod)delay = cost / quickfor i in range(quick):await anyio.sleep(delay)progress.advance(task)cost = seconds - costslow = total - quickdelay = cost / slowfor i in range(slow):await anyio.sleep(delay)progress.advance(task)with Progress(transient=transient) as progress:task = progress.add_task(f"[{color}]{msg}:", total=total)async with anyio.create_task_group() as tg:tg.start_soon(play, progress, task)yieldtg.cancel_scope.cancel()progress.update(task, completed=total)@asynctor.timeit
async def fetch(url: str, timeout: int = 5) -> int:async with httpx.AsyncClient(timeout=timeout, verify=False, follow_redirects=True) as client:r = await client.get(url)return r.status_codeasync def main() -> None:url = "https://pypi.org/search/?q=asynctor"async with percentbar(f"Fetching {url}"):status_code = await fetch(url)print(f"{status_code = }")url = "https://fastapi.tiangolo.com/"async with percentbar(f"Fetching {url}", seconds=8, color="green", transient=True):status_code = await fetch(url, timeout=8)print(f"{url = }; {status_code = }")if __name__ == "__main__":asynctor.run(main)

效果如下：