Multistatus for asyncio tasks/gather.

multi_status.py

from rich.console import Console, Group
from rich.live import Live
from rich.status import Status
import asyncio
import time

class MultiStatus:
    def __init__(self, tasks):
        self.statuses = {
            name: Status(f"{name}: Starting...", spinner="dots", spinner_style="green")
            for name in tasks
        }
        
    def update_task(self, task_name: str, message: str):
        self.statuses[task_name].update(f"{task_name}: {message}")
                
    def __rich__(self):
        # Create a group of renderable status objects
        return Group(*self.statuses.values())

async def monitored_task(name: str, duration: float, multi_status: MultiStatus):
    start = time.time()
    while (time.time() - start) < duration:
        progress = int((time.time() - start) / duration * 100)
        multi_status.update_task(name, f"Processing... {progress}%")
        await asyncio.sleep(0.1)
    
    multi_status.update_task(name, "✓ Done!")
    return f"Result from {name}"

async def run_tasks_with_status():
    console = Console()
    
    # Create tasks list and status
    task_names = ["Task 1", "Task 2", "Task 3"]
    multi_status = MultiStatus(task_names)
    
    # Create tasks
    tasks = [
        monitored_task("Task 1", 3.0, multi_status),
        monitored_task("Task 2", 5.0, multi_status),
        monitored_task("Task 3", 4.0, multi_status)
    ]
    
    # Run everything with live display
    with Live(multi_status, refresh_per_second=10, transient=False):
        results = await asyncio.gather(*tasks)
    
    # Print final results
    console.print("\nAll tasks completed!")
    for name, result in zip(task_names, results):
        console.print(f"{name} result: {result}")

if __name__ == "__main__":
    asyncio.run(run_tasks_with_status())