A progress bar for sequential work is straightforward: one item at a time, advance after each. Concurrent work breaks the picture. Eight uploads run at once, each at its own speed; some finish in a second, one takes a minute, two fail halfway. A single bar that jumps when items complete hides which files are in flight and why the last 10% takes forever. A bar per item, all forty of them, floods the screen. And none of it should appear when the output goes to a CI log, where animated bars become thousands of lines of escape codes. This guide builds a progress display for concurrent tasks that shows the right things: one overall bar with counts and an ETA, one transient row per in-flight item that disappears when the item finishes, failures printed permanently above the live area as they happen, a final summary, and plain periodic lines instead of animation when stdout is not a terminal. It belongs to the concurrency and async topic.
Prerequisites
uv add rich(examples checked with Rich 15 and Python 3.13).- Adding progress bars and spinners to Python CLIs for Rich progress basics, and structured concurrency with TaskGroups for the concurrency side.
What to show
The layout answers the questions a user has while waiting, in order of importance. How far along is it, and how long will it take? — the overall bar with “23/40” and a time estimate. Is it stuck? — the in-flight rows, each with its own small bar and byte count, which show that work is moving and which item is slow. Did anything go wrong? — failure lines that stay on screen after the live area is gone, so the user does not have to scroll through a log at the end. The number of in-flight rows equals the concurrency limit, so the display has a fixed height no matter how many items there are.
The recipe
# src/mytool/concurrent_progress.py
from __future__ import annotations
import asyncio
import time
from collections.abc import Awaitable, Callable, Sequence
from rich.console import Console, Group
from rich.live import Live
from rich.progress import (BarColumn, DownloadColumn, MofNCompleteColumn, Progress, SpinnerColumn,
TaskID, TextColumn, TimeRemainingColumn)
class ConcurrentProgress:
"""One overall bar plus a row per in-flight item; plain periodic lines when not a terminal."""
def __init__(self, total: int, console: Console, *, label: str = "items",
plain_every: float = 5.0) -> None:
self.total, self.label, self.console = total, label, console
self.done = self.failed = 0
self.interactive = console.is_terminal
self.plain_every, self._last_plain = plain_every, time.monotonic()
self.progress = Progress(
SpinnerColumn(), TextColumn("{task.description}", style="bold"),
BarColumn(), MofNCompleteColumn(), TimeRemainingColumn(),
console=console, transient=False, disable=not self.interactive,
)
self.items = Progress(
TextColumn(" {task.description}"), BarColumn(bar_width=20), DownloadColumn(),
console=console, transient=True, disable=not self.interactive,
)
self.overall: TaskID = self.progress.add_task(label, total=total)
def __enter__(self) -> ConcurrentProgress:
if self.interactive:
self.live = Live(Group(self.progress, self.items), console=self.console,
refresh_per_second=10, transient=False)
self.live.__enter__()
return self
def __exit__(self, *exc) -> None:
if self.interactive:
self.live.__exit__(*exc)
self.console.print(f"{self.done - self.failed} of {self.total} {self.label} done"
+ (f", {self.failed} failed" if self.failed else ""))
def start_item(self, name: str, size: int | None) -> TaskID:
return self.items.add_task(name, total=size)
def advance_item(self, item: TaskID, amount: int) -> None:
self.items.update(item, advance=amount)
def finish_item(self, item: TaskID, name: str, error: str | None = None) -> None:
self.items.remove_task(item)
self.done += 1
if error:
self.failed += 1
self.console.print(f"[red]✗[/red] {name}: {error}") # stays above the live area
self.progress.update(self.overall, advance=1,
description=f"{self.label} ({self.failed} failed)" if self.failed else self.label)
self._maybe_plain()
def _maybe_plain(self) -> None:
now = time.monotonic()
if not self.interactive and (now - self._last_plain >= self.plain_every or self.done == self.total):
self._last_plain = now
self.console.print(f"progress: {self.done}/{self.total} {self.label}, {self.failed} failed")
async def run_all(items: Sequence[tuple[str, int]], worker: Callable[..., Awaitable[None]],
console: Console, *, jobs: int = 4) -> ConcurrentProgress:
gate = asyncio.Semaphore(jobs)
with ConcurrentProgress(len(items), console, label="files") as ui:
async def one(name: str, size: int) -> None:
async with gate:
row = ui.start_item(name, size)
try:
await worker(name, size, lambda n: ui.advance_item(row, n))
except OSError as exc:
ui.finish_item(row, name, str(exc))
else:
ui.finish_item(row, name)
async with asyncio.TaskGroup() as tg:
for name, size in items:
tg.create_task(one(name, size))
return ui
Two Progress objects in one Live display
Rich’s Progress renders all of its tasks with the same columns, but the overall bar and the per-item rows need different columns: counts and ETA for the first, bytes for the second. The recipe uses two Progress instances and combines them in a Group inside a single Live display, which owns the terminal and refreshes both at ten frames per second. The item Progress is transient, and rows are removed with remove_task the moment an item finishes, so only in-flight work is visible.
Rows appear only while work is running
The worker acquires the semaphore before adding its row. Items waiting for a slot have no row, so the in-flight list never shows forty idle entries — just the eight that are actually transferring. The byte-level advance callback is passed down to the worker, which calls it as chunks are sent; how a worker reports bytes from an HTTP client is shown in downloading files with progress in Python.
Failures go above the live area
console.print called while a Live display is active prints above it, and the line stays in the scrollback when the display ends. That makes it the right place for failures — “✗ file-3.bin: connection reset” — as they happen, instead of collecting them for the end. The overall bar’s description also changes to “files (1 failed)” so the count is visible at a glance.
Pipes and CI: plain lines, not animation
When the console is not a terminal, both Progress objects are created with disable=True and no Live display starts. Instead, _maybe_plain prints a short status line at most every five seconds and once at the end: progress: 23/40 files, 1 failed. CI logs get a readable record of how the run progressed, with no escape codes and no thousands of redraw lines; failure lines and the final summary are printed in both modes. The same principle is discussed in detecting CI environments and non-interactive shells.
UX considerations
- Progress belongs on stderr. Pass a
Console(stderr=True)so stdout stays free for results, as in separating logs from program output. - Estimate from bytes when sizes vary. An ETA based on item counts is wrong when one item is a 2 GB file; if sizes are known up front, give the overall bar a byte total instead.
- Keep refreshes modest. Ten refreshes per second is smooth; updating per chunk from hundreds of tasks is fine because Rich renders on its own schedule, not on every update.
- Do not hide the slow item. If one transfer stalls, its row staying put while others churn is exactly the information the user needs.
- Finish with a summary line, including failures and a hint how to retry only those (“retry with
--only-failed”).
Testing the behaviour
Drive the display with fake workers and two consoles writing to StringIO: one forced into terminal mode to exercise the live display, one plain to exercise the pipe path:
# tests/test_concurrent_progress.py
import asyncio
import io
from rich.console import Console
from mytool.concurrent_progress import run_all
ITEMS = [(f"file-{n}.bin", 4096) for n in range(10)]
async def fake_upload(name, size, advance):
for _ in range(4):
await asyncio.sleep(0.001)
advance(size // 4)
if name == "file-3.bin":
raise OSError("connection reset")
def console(terminal: bool) -> Console:
return Console(file=io.StringIO(), force_terminal=terminal, width=80, color_system=None)
def test_counts_and_summary_in_a_pipe():
out = console(terminal=False)
ui = asyncio.run(run_all(ITEMS, fake_upload, out, jobs=3))
assert (ui.done, ui.failed) == (10, 1)
text = out.file.getvalue()
assert "✗ file-3.bin: connection reset" in text
assert text.rstrip().endswith("9 of 10 files done, 1 failed")
assert "progress: 10/10 files, 1 failed" in text # final plain progress line
assert "\x1b[" not in text # no live-display escape codes
def test_live_display_on_a_terminal():
out = console(terminal=True)
ui = asyncio.run(run_all(ITEMS, fake_upload, out, jobs=3))
text = out.file.getvalue()
assert ui.done == 10
assert "10/10" in text and "files (1 failed)" in text
assert "progress:" not in text # plain lines only without a TTY
Assertions focus on what the user must be able to rely on — counts, the failure line, the final summary, no escape codes in plain mode, no plain progress lines in terminal mode — rather than the exact frames, which depend on timing. The fake worker’s tiny sleeps keep the suite fast while still interleaving the tasks.
Conclusion
Concurrent work needs a progress display with structure: an overall bar with counts and ETA, a transient row per in-flight item created after the concurrency slot is acquired, failures printed above the live area as they happen, and a summary at the end. Combine two Progress instances in one Live group, send it to stderr, switch to periodic plain lines when the console is not a terminal, and test both modes with fake workers and string-backed consoles.
Frequently asked questions
Is Rich’s Progress safe to update from many asyncio tasks?
Yes. Asyncio tasks run in one thread, and Progress methods are also protected by a lock, so updates from threads work too — which matters when work runs in a thread pool.
How do I show progress for thread-pool work?
The same class works: call start_item, advance_item and finish_item from worker threads. Live refreshes from its own thread, so the display stays smooth while workers block. See parallelising CLI work with thread pools.
What if item sizes are unknown?
Pass total=None for the row; Rich then shows a pulsing bar and the byte count without a percentage. The overall bar still counts items.
Should verbose mode change the display?
At -v, also print a line for each successful item above the live area. Without it, only failures are printed, which keeps the scrollback short.
Can I use tqdm instead?
tqdm supports multiple bars with position=, and works well for a fixed number of workers. Rich’s Live group makes the dynamic rows and printing above the display easier, and fits better if the CLI already uses Rich.