Parallel command invocation with specified workers

parallel.py

import argparse as ap
import dataclasses as dc
import multiprocessing as mp
import subprocess as sp
import sys


def parse_args() -> ap.Namespace:
    p = ap.ArgumentParser("Runs parallel jobs")

    p.add_argument("-j", "--njobs", type=int, default=1, help="Number of jobs")
    p.add_argument("-v", "--verbose", action="store_true", help="Whether to print incoming commands")
    p.add_argument("commands", type=str, nargs="+", help="Commands to run")

    args = p.parse_args()
    if not args.njobs >= 1:
        raise ValueError("njobs must be 1 or greater.")
    return args


@dc.dataclass(frozen=True)
class ProcessArgs:
    command: str
    verbose: bool


def process(args: ProcessArgs) -> None:
    if args.verbose:
        print(f"parallel.py: Running: {args.command}", file=sys.stderr)
    sp.run(args.command, shell=True)


def main() -> None:
    args = parse_args()
    pool = mp.Pool(args.njobs)
    proc_args = (ProcessArgs(command=c, verbose=args.verbose) for c in args.commands)
    for _ in pool.imap_unordered(process, proc_args):
        pass
    


if __name__ == "__main__":
    main()