rich+asyncio

[rendarz]

What about using rich.live with asyncio? This code miserably fails, because it instances Live() more than once. -> https://bpa.st/2FZA

My aim here, is a very simple thing, that a library like rich should easily let to do:

• An asyncio coroutine print "Waiting ..." then it starts the job (e.g. HTTP request) and when finished prints "Done."

1. Let's say we call N of those coroutines with asyncio.gather().
2. That coroutine (called N times) print "Waiting ..." N times, and each of them starts his job.
3. Other part of the program prints other lines into the console, so the "Waiting ..." lines goes up.
4. Each coroutine, when it finishes its job, prints "Done." but that string goes in the row where the coroutine initially printed "Waiting ...." and not in current console new lines.

This is like having N progress bars that are printed in the console, while also other text is printed in the console, and those progress bars still continue their job.
In my case we don't have a progress bar, but a much simpler thing: print("Waiting .........") and then print("Done.") in the same line.

Is this possible with current rich capabilities? It would be very useful to support such as behaviors within asyncio!

[willmcgugan]

You can use Live with asyncio, but you would structure it differently. Have a single instance of Live running in a thread. Your tasks can run in another thread and call live.update when there is a new table.

You could also use a Progress class which has support for multiple jobs (you could replace the bars with a spinner).

[rendarz]

Hi, thanks for your reply, and your effort for this awesome library which helped many people!
Well, I have done some tests, and the best I could have get is this:

import sys
import asyncio
from rich.live import Live
from rich.progress import Progress

ADD_JOBS = True

async def do_task(name: str, progress, task, sleep) -> None:
    await asyncio.sleep(sleep)
    progress.update(task, description=f"[green]DONE! {name}", advance=100.0)

progress = Progress()
live = Live(progress, refresh_per_second=4)

async def add_jobs(i:int) -> int:
  sleeps = [2.0, 3.0, 1.0, 0.5, 3.0, 4.0]
  jobs = [f"MIAO{i}", f"BAU{i}", f"Hello{i}", f"world{i}", f"mais{i}", f"USA{i}"]
  coros = []
  for job, sleep in zip(jobs, sleeps):
    task = progress.add_task(f"[red]Downloading... {job}", total=100)
    coros.append(do_task(job, progress, task, sleep))
  with live:
    await asyncio.gather(*coros)
  return 0


async def other_jobs():
  for i in range(15):
    print(f"Doing also job {i} ...")
    await asyncio.sleep(0.5)
    if ADD_JOBS and (i == 5):
      at = asyncio.create_task(add_jobs(2))
  await at

async def run_all():
  await asyncio.gather(other_jobs(), add_jobs(0), add_jobs(1))


def main() -> int:
  return asyncio.run(run_all())

if __name__ == "__main__":
  sys.exit(main())

This is the best I could, but it still doesn't work: If you run this, some tasks will remain incomplete.
I will attach the screen of my console. Why that happens? Is that a bug?

Thank you.

[graingert]

remember to await your async functions as soon as you call them:

import sys
import anyio
import asyncio
from rich.console import RichCast
from rich.live import Live
from rich.progress import Progress, TaskID

ADD_JOBS = True


async def do_task(name: str, progress: Progress, task: TaskID, sleep: float) -> None:
    await asyncio.sleep(sleep)
    progress.update(task, description=f"[green]DONE! {name}", advance=100.0)


async def add_jobs(progress: Progress, live: Live, i: int) -> int:
    sleeps = [2.0, 3.0, 1.0, 0.5, 3.0, 4.0]
    jobs = [f"MIAO{i}", f"BAU{i}", f"Hello{i}", f"world{i}", f"mais{i}", f"USA{i}"]

    async def go(job: str, sleep: float) -> None:
        task = progress.add_task(f"[red]Downloading... {job}", total=100)
        await do_task(job, progress, task, sleep)

    await asyncio.gather(*(go(job, sleep) for job, sleep in zip(jobs, sleeps)))
    return 0


async def other_jobs(progress: Progress, live: Live) -> None:
    async with anyio.create_task_group() as tg:
        for i in range(15):
            print(f"Doing also job {i} ...")
            await asyncio.sleep(0.5)
            if ADD_JOBS and (i == 5):
                tg.start_soon(add_jobs, progress, live, 2)


async def amain() -> int:
    progress = Progress()
    rich_progress: RichCast = progress  # type: ignore[assignment]
    with Live(rich_progress, refresh_per_second=4) as live:
        await asyncio.gather(
            other_jobs(progress, live),
            add_jobs(progress, live, 0),
            add_jobs(progress, live, 1),
        )
    return 0


def main() -> int:
    return asyncio.run(amain())


if __name__ == "__main__":
    sys.exit(main())

[graingert]

something like this?

import sys
import asyncio
from rich.live import Live
from rich.table import Table


async def do_task(name: str, live: Live, table: Table) -> None:
    for row in range(12):
        live.console.print(f"Working on row #{row} {name}")
        await asyncio.sleep(2.0)
        table.add_row(f"{row}", f"description {row} {name}", "[red]ERROR")


async def amain() -> int:
    table = Table()
    table.add_column("Row ID")
    table.add_column("Description")
    table.add_column("Level")

    with Live(
        table, refresh_per_second=4
    ) as live:  # update 4 times a second to feel fluid
        await asyncio.gather(do_task("BAU", live, table), do_task("MIAO", live, table))

    return 0


def main() -> int:
    return asyncio.run(amain())


if __name__ == "__main__":
    sys.exit(main())