Function run_parallel

misc/python/materialize/parallel_task.py:131–183

Run tasks in parallel and print results as each task finishes. Args: tasks: List of (name, spec) pairs. verbose: If True, print output even for successful tasks. print_duration: If True, include duration in status lines. spinner_suffix: Suffix after the count in

(
    tasks: list[tuple[str, TaskSpec]],
    verbose: bool = False,
    print_duration: bool = True,
    spinner_suffix: str = "tasks",
    print_summary: bool = True,
)

def run_parallel(
    tasks: list[tuple[str, TaskSpec]],
    verbose: bool = False,
    print_duration: bool = True,
    spinner_suffix: str = "tasks",
    print_summary: bool = True,
) -> list[str]:
    """Run tasks in parallel and print results as each task finishes.

    Args:
        tasks: List of (name, spec) pairs.
        verbose: If True, print output even for successful tasks.
        print_duration: If True, include duration in status lines.
        spinner_suffix: Suffix after the count in the spinner, e.g. "tasks".
        print_summary: If True, print a final success/failure summary.

    Returns:
        List of failed task names (empty on full success).
    """
    done_q: queue.Queue[TaskThread] = queue.Queue()
    threads = [TaskThread(name, spec, done_queue=done_q) for name, spec in tasks]

    spinner = _SpinnerThread(len(threads), spinner_suffix)
    spinner.start()

    for t in threads:
        t.start()

    failed = []
    for _ in threads:
        t = done_q.get()
        spinner.task_done()
        spinner.clear_line()
        formatted_duration = (
            f" [{t.duration.total_seconds():5.2f}s]" if print_duration else ""
        )
        if t.success:
            print(f"{_prefix('---')}{OK}{formatted_duration} {t.name}")
        else:
            print(f"{_prefix('+++')}{FAIL}{formatted_duration} {t.name}")
            failed.append(t.name)
        if t.output and (not t.success or verbose):
            print(t.output)

    spinner.stop()

    if print_summary:
        if failed:
            print(f"{_prefix('+++')}{FAIL} Failed: {failed}")
        else:
            print(f"{_prefix('+++')}{OK} All tasks successful")

    return failed