Skip to content

Loading big data fixtures: Showing progress

Recently I felt the need to show progress during a multiprocessing command loading a large set of fixtures

Here a slimmed down version without the data loading itself.

The command is executed with the help of typer

import time
from multiprocessing import Pool
from rich.progress import Progress
import random
import typer

def worker(_: int) -> None:
    time.sleep(random.randint(0, 5) / 10)

def main(processes: int = 4) -> None:
    tasks = range(100)
    with Progress() as progress:
        task_id = progress.add_task("[cyan]Completed...", total=len(tasks))
        with Pool(processes) as pool:
            results = pool.imap(
                worker,
                tasks
            )
            for _ in results:
                progress.advance(task_id)


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