How to build a parallel job runner in Python, one library at a time

Say you want a small tool that reads a list of shell jobs from a config file, runs them in parallel with a limit on how many go at once, and prints a JSON report of what passed and what failed, exiting non-zero if anything failed.

Each library involved (subprocess, concurrent.futures, PyYAML, json) is documented well on its own. What's hard to find is how the pieces fit, and why each piece is shaped the way it is. Every example is either a one-liner or a full framework. This is the middle ground.

Build it in seven steps. Each step adds exactly one library and solves exactly one problem. If you understand why each step exists, you can rebuild the whole thing from scratch instead of remembering it.

Step 1: run one job

The only call you need from subprocess:

import subprocess

proc = subprocess.run("echo hello", shell=True, capture_output=True, text=True)

proc.returncode   # 0
proc.stdout       # "hello\n"
proc.stderr       # ""

Four keywords, four reasons:

• shell=True lets you pass one string and get pipes, globs, and &&. Without it you must pass a list like ["echo", "hello"], and "echo hello" fails with FileNotFoundError because it looks for a program literally named echo hello.
• capture_output=True puts the output on the result object. Without it, output goes straight to your terminal and you cannot inspect it.
• text=True decodes to str. Without it you get bytes, and proc.stdout == "hello\n" is silently False forever.
• No check=True. A failing job is data, not an exception. You want to record it and keep going, so you read .returncode yourself.

Wrap it so it returns a plain dict rather than a library object. This matters later: dicts serialize to JSON, CompletedProcess does not.

def run_command(job):
    proc = subprocess.run(job["command"], shell=True,
                          capture_output=True, text=True)
    return {
        "status": "ok" if proc.returncode == 0 else "failed",
        "exit_code": proc.returncode,
        "stdout": proc.stdout,
        "stderr": proc.stderr,
    }

The "ok" if ... else "failed" string is doing real work: it becomes the thing you filter on at the end, and it is readable in the JSON output. An exit code alone forces every consumer to know that 0 means success.

Step 2: run them all, one at a time

report = {}
for name, job in commands.items():
    report[name] = run_command(job)

This is correct and slow. Everything that follows exists only to make it faster without changing what it produces. Keep this version in your head as the definition of correct behaviour.

Step 3: run them in parallel with a cap

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=4) as pool:
    ...

Two things to understand here.

Why threads and not processes. The usual warning is that the GIL prevents threads from running Python in parallel. That is true and irrelevant here, because your threads are not running Python. They are blocked inside subprocess.run waiting on an operating system call, and Python releases the GIL while waiting. Threads are correct for I/O and for waiting on child processes. Use ProcessPoolExecutor only when the Python code itself is the expensive part.

Why the with block. Exiting it calls pool.shutdown(wait=True), which blocks until everything already submitted has finished. It does not cancel anything. This is usually what you want, but it means an exception escaping the block does not stop the work, it just delays the traceback until the work is done.

max_workers is the whole concurrency cap. You do not have to ration submissions yourself: submit as much as you like and the executor queues the excess.

Step 4: react as each job finishes

This is the step people find genuinely confusing, so go slowly.

future = pool.submit(run_command, job)

submit returns immediately, before the job has run. What you get back is a Future: a handle to a result that does not exist yet. Nothing blocks until you ask for the value.

Two problems follow from that, and each has a standard answer.

Problem 1: the Future does not know what it is. There is no future.args. Once you have a finished future, you have no way to tell which job it was, and completion order is not submission order. The fix is to record the association yourself. Futures are hashable, so use them as dict keys:

in_flight = {}
in_flight[pool.submit(run_command, commands[name])] = name

Read that inside-out: submit the job, get a future, store future -> name. It looks strange the first time. It is the standard idiom.

Problem 2: knowing when something has finished. You could loop and poll future.done(). Do not. Use wait:

from concurrent.futures import wait, FIRST_COMPLETED

done, not_done = wait(in_flight, return_when=FIRST_COMPLETED)

• It returns a two-tuple of sets. Write done, _ = wait(...) when you do not need the second half.
• It accepts any iterable of futures. Passing the dict works because iterating a dict yields its keys, which are the futures.
• return_when=FIRST_COMPLETED returns as soon as at least one is done. The default is ALL_COMPLETED, which waits for everything. Omitting this is the single easiest mistake to make here, because the program still works, it just stops being incremental.

Now drain the finished ones:

for fut in done:
    name = in_flight.pop(fut)       # pop, so it leaves the in-flight set
    report[name] = fut.result()

fut.result() returns the worker's return value, and re-raises whatever the worker raised. A worker that blew up still appears in done and still looks complete. If you never call .result() or .exception(), that error vanishes without a trace. If you want one bad job not to kill the run:

    try:
        report[name] = fut.result()
    except Exception as exc:
        report[name] = {"status": "error", "error": str(exc)}

Putting step 4 together. If your work list is fixed, as_completed is shorter and you should use it:

from concurrent.futures import as_completed

with ThreadPoolExecutor(max_workers=4) as pool:
    futures = {pool.submit(run_command, job): name for name, job in commands.items()}
    for fut in as_completed(futures):
        report[futures[fut]] = fut.result()

If finishing one job can make new jobs available, as_completed does not work, because it iterates a snapshot taken when you called it. You need an explicit loop:

with ThreadPoolExecutor(max_workers=4) as pool:
    while pending or in_flight:
        while pending:                                   # submit everything available
            name = pending.pop()
            in_flight[pool.submit(run_command, commands[name])] = name

        done, _ = wait(in_flight, return_when=FIRST_COMPLETED)

        for fut in done:
            name = in_flight.pop(fut)
            report[name] = fut.result()
            pending.extend(dependents_of(name))            # may be empty

while pending or in_flight is the termination condition and it is load-bearing. Never call wait() on an empty collection inside a loop: it returns instantly and you spin at 100% CPU.

One thing that surprises people: report, pending, and in_flight need no locks. Only the main thread touches them. The worker threads receive arguments and return values and never share state. Structuring it this way is why you can skip the entire topic of thread safety.

Step 5: read the job list from a file

import yaml

with open(config_path) as f:
    data = yaml.safe_load(f)

• Always safe_load, never load. Plain yaml.load can construct arbitrary Python objects, which is remote code execution on a file you did not write.
• An empty file parses to None, not {}. data["commands"] then raises TypeError, which sends you hunting in the wrong place. Check isinstance(data, dict) first.
• Under YAML 1.1, unquoted yes, no, on, and off become booleans.
• PyYAML is third party despite feeling standard: pip install pyyaml, import as yaml.

Step 6: report the report

import json
from datetime import datetime, timezone

def utc_stamp():
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

Why this exact line:

• datetime.now(timezone.utc) gives an aware datetime. datetime.utcnow() gives a naive one holding UTC values, so it raises TypeError the moment you compare it with an aware datetime. It is deprecated as of 3.12.
• .strftime(...) rather than .isoformat() because isoformat on an aware UTC datetime produces +00:00, and most consumers expect the Z suffix.
• It returns a string, and that is the real reason this helper exists: json.dumps cannot serialize a datetime. Converting at the boundary means you never think about it again.

summary = {
    "started": started,        # captured before the work
    "ended": utc_stamp(),        # captured after
    "report": report,
}
print(json.dumps(summary, indent=2))

• The s suffix means string. dumps/loads work with strings, dump/load work with file objects. You want dumps here because you are printing.
• indent=2 pretty-prints. Without it you get one long line.
• sort_keys=True is worth adding if you ever diff the output.
• Not serializable: datetime, set, bytes, Decimal. The escape hatch is default=str, but converting deliberately is better.

Send this to stdout and nothing else. Diagnostics go to stderr:

print("could not read config", file=sys.stderr)

If you interleave them, nobody can pipe your output into jq.

Step 7: the exit code

failed = any(r["status"] != "ok" for r in report.values())
return 1 if failed else 0

any() short-circuits on the first match and takes a generator expression, so nothing is materialized. Note != "ok" rather than == "failed": if you later add a "timeout" or "error" status, the first version keeps working and the second silently reports success.

Return an int from your function and convert it at the very bottom:

if __name__ == "__main__":
    sys.exit(main())

Returning rather than calling sys.exit() in the middle of your logic is what makes the whole thing testable: a test calls main([...]) and asserts on the returned int.

The shape to remember

Once you have the steps, the program is just this order:

1. Parse arguments
2. Load and validate config, return non-zero on bad input
3. Timestamp the start
4. Open a pool, submit work, collect report as they finish
5. Timestamp the end
6. print(json.dumps(summary, indent=2)) to stdout
7. Return 0 or 1 based on the report

Steps 1 through 4 are the only part with any real subtlety, and within those, three details account for most of the bugs: text=True on subprocess.run, return_when=FIRST_COMPLETED on wait, and calling .result() so that worker exceptions actually surface.

Comments

[Raknaos]

Two lines here deserve to be on a poster: never call wait() on an empty collection inside the loop because it returns instantly and you spin at 100% CPU, and a worker that blew up still shows up in done and still looks complete unless you touch .result() or .exception(). Both fail without raising anything, which is exactly why they survive review.

The as_completed versus wait fork is the part most writeups skip. as_completed iterates a snapshot taken when you call it, so the moment finishing one job can enqueue another one it is silently wrong and you are back to an explicit loop with pending and in_flight. Good to see the termination condition named as load-bearing rather than decoration — and the isinstance(data, dict) check before indexing YAML earns its place, since an empty config failing as a TypeError several lines later sends you hunting in the wrong file.