In order to show a progress bar, Rich needs to know how may steps are involved and when you finish a step. The track function can get this information automatically from a sequence. You're using this in your example, but your sequence only has a single element so you go from 0 to 100% in a single step.
If you want to track progress of something you need a sequence that defines the work to be done. For instance if you had a list of urls to scrape, you might do something like this:
from rich.progress import track
SCRAPE_URLS = ["https://example.org", "https://google.org", ...]
for url in track(SCRAPE_URLS):
scrape(url)
The progress bar will advance for every URL.
Answer from Will McGugan on Stack OverflowCan't get progress bar to work in python rich - Stack Overflow
Displaying an "indeterminate" progress bar in the terminal with Rich.
multithreading - How to create individual rich progress bars for each worker in Python multiprocessing's imap_unordered()? - Stack Overflow
Progress bar fixed to bottom of window
In order to show a progress bar, Rich needs to know how may steps are involved and when you finish a step. The track function can get this information automatically from a sequence. You're using this in your example, but your sequence only has a single element so you go from 0 to 100% in a single step.
If you want to track progress of something you need a sequence that defines the work to be done. For instance if you had a list of urls to scrape, you might do something like this:
from rich.progress import track
SCRAPE_URLS = ["https://example.org", "https://google.org", ...]
for url in track(SCRAPE_URLS):
scrape(url)
The progress bar will advance for every URL.
In order for the rich package to work properly, you need to check Emulate terminal in output console, which you can find in the Run menu (on top), then Edit configurations, then Modify options.

I had your same problem (the bar appeared only after the process was done), and this solved it.
I heavily borrowed from @Gothboidesu previous answer and mostly changed it so that the workers have pulsing bars (each worker's total number can't be determined beforehand) and only the main bar having a progress %.

import random
import time
import threading
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeRemainingColumn
from rich.live import Live
from rich.console import Console
from multiprocessing.dummy import Pool
def do_something(number):
thread_name = threading.current_thread().name
with task_lock:
if thread_name not in thread_task_map:
# Create a new bar for this thread
thread_task_map[thread_name] = progress.add_task(f"[green]{thread_name}", total=None, visible=True)
task = thread_task_map[thread_name]
time.sleep(number/100)
progress.update(task, advance=1)
progress.update(overall_task, advance=1)
numbers = random.sample(range(1, 101), 50)
pool = Pool(4)
console = Console()
progress = Progress(
SpinnerColumn(),
TextColumn("{task.description}"),
BarColumn(),
TextColumn("{task.completed} numbers crunched."),
TimeRemainingColumn()
)
# Dictionary to hold a task per thread
thread_task_map = {}
task_lock = threading.Lock()
# Overall progress bar
overall_task = progress.add_task("[cyan]Overall", total=len(numbers))
with Live(progress, refresh_per_second=30, console=console):
pool.imap_unordered(do_something, numbers)
pool.close()
pool.join()
A multithreading pool uses a single input task queue and when a thread becomes idle it takes from that queue the next task to execute. But you have no real control over which thread will process which task and how many total tasks each task will process. So how would you know how many total tasks each thread would eventually handle?
This solution creates individual threads and splits up the input tasks into N_THREADS chunks and passes one of the chunks to each of the threads. Optionally, a threading.Queue instance can be passed to the threads if a result value needs to be returned:
import time
import random
from threading import Thread
from queue import Queue, Empty
from typing import Callable, Optional, Any, Generator
from rich.progress import Progress
from rich.live import Live
N_THREADS = 4
class WorkerThread:
"""This class encapsulates a Thread and initializes a progress bar."""
def __init__(self, args: list[any], worker: Callable[[any], any], out_q: Optional[Queue]=None):
# Each element of args is a new task to be executed by this thread and it can
# be of any type. If the worker function requires multiple input values, then
# each element could be, for example, a tuple that gets unpacked by the worker function
# or perhaps a dictionary of named arguments.
self._args = args
self._out_q = out_q
self._worker = worker
self._thread = Thread(target=self.run)
self._task = progress.add_task(self._thread.name, total=len(self._args))
def start(self):
self._thread.start()
def join(self) -> None:
self._thread.join()
def run(self) -> None:
for arg in self._args:
result = self._worker(arg)
if self._out_q:
# return result:
self._out_q.put(result)
progress.update(self._task, advance=1)
progress.update(master_task, advance=1)
def do_something(number):
# slow down the progress a bit to better see what is happening:
time.sleep(number/25)
return number
def chunks(lst: list[Any], number_chunks: int) -> Generator[Any, None, None]:
list_size = len(lst)
chunk_size = list_size // number_chunks
offset = 0
for i in range(number_chunks - 1):
yield lst[offset:offset+chunk_size]
offset += chunk_size
yield lst[offset:]
def main():
global progress, master_task
N_NUMBERS = 50
numbers = random.sample(range(1, 101), N_NUMBERS)
out_q = Queue()
progress = Progress()
with Live(progress):
master_task = progress.add_task('All tasks', total=N_NUMBERS)
args_list = list(chunks(numbers, N_THREADS))
worker_threads = [WorkerThread(args, do_something, out_q) for args in args_list]
for worker_thread in worker_threads:
worker_thread.start()
for worker_thread in worker_threads:
worker_thread.join()
try:
results = []
while True:
results.append(out_q.get_nowait())
except Empty:
pass
#print(results)
if __name__ == '__main__':
main()