python之七彩进度条rich.progress

Spinner

中国有七彩云南，Python有七彩进度条rich.progress.Progress

#!/usr/bin/env python
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 spinnerbar(
    msg: str, color: str = "cyan", transient=True
) -> 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_code


async def main() -> None:
    url = "https://pypi.org"
    with spinnerbar(f"Fetching {url}", transient=False):
        status_code = await asynctor.timeit(fetch)(url)
    print(f"{status_code = }")
    url = "https://fastapi.tiangolo.com/"
    with spinnerbar(f"Fetching {url}", color="green"):
        status_code = await fetch(url)
    print(f"{url = }; {status_code = }")


if __name__ == "__main__":
    asynctor.run(main)

虽然看不出具体进度，但起码知道程序正在运行，而不是卡死或挂掉：

Percent Bar

• main.py

#!/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 * 100

    async def play(progress, task, expected=1 / 2, thod=0.8):
        # 改变进度条速率，前面1/2的时间完成80%的进度
        cost = seconds * expected
        quick = int(total * thod)
        delay = cost / quick
        for i in range(quick):
            await anyio.sleep(delay)
            progress.advance(task)
        cost = seconds - cost
        slow = total - quick
        delay = cost / slow
        for 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)
            yield
            tg.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_code


async def main() -> None:
    url = "https://pypi.org/search/?q=asynctor"
    async with percentbar(f"Fetching {url}"):
        status_code = await asynctor.timeit(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)

• Result

这个进展虽然不一定正确反映任务进度，但起码让人知道任务正在向前推进着，也大概知道什么时候能完成。