Parallelising CLI Work with Thread Pools

Your command loops over a list — files to upload, hosts to check, repositories to clone, API records to fetch — and each iteration spends most of its time waiting on the network or disk. With 400 items at 300 ms each, the user waits two minutes while the CPU sits idle. A thread pool is the smallest change that fixes it: your existing synchronous code runs in several threads at once, and blocking I/O releases Python's global interpreter lock so the waits overlap. Done carelessly, though, it produces interleaved output, loses exceptions, ignores Ctrl+C and hammers whatever it talks to. This guide builds a thread-pool pattern for CLI commands that is bounded, reports progress, collects every failure, stops cleanly and prints deterministic results. It is part of the concurrency and async topic.

Prerequisites

• Python 3.10+, Typer and Rich.
• Work that is I/O-bound: network calls, subprocesses, file operations. For CPU-bound Python, see multiprocessing for CPU-bound CLI tasks instead.
• Thread-safe clients for anything shared between workers. httpx.Client, logging and Rich consoles are; many SDK clients and database connections are not.

How a pool works

A ThreadPoolExecutor starts a fixed number of worker threads and a queue. Each submit() puts a call on the queue and returns a Future immediately; workers pull calls off the queue and run them; the future completes with the return value or the exception.

The pool size is your concurrency limit: with max_workers=8, at most eight items are in progress at any moment, however many you submit. That is the property that keeps you from opening 400 connections at once.

There are two ways to get results back, and for CLIs the choice matters:

executor.map(fn, items) is a one-liner that yields results in input order. If item 3 is slow, results for items 4–400 wait behind it, so a progress bar jumps in bursts; and the first exception is raised when the iteration reaches it, abandoning the rest. as_completed(futures) yields futures as they finish, which gives smooth progress and lets you handle each failure individually. It is the better default for commands.

The recipe

The example command uploads every file in a directory to an HTTP endpoint. The same structure fits any per-item I/O task.

# src/mytool/upload.py
from __future__ import annotations

import threading
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from pathlib import Path

import httpx
from rich.progress import BarColumn, MofNCompleteColumn, Progress, TextColumn, TimeElapsedColumn


@dataclass(frozen=True)
class Result:
    path: Path
    status: int | None = None
    error: str | None = None


def upload_one(client: httpx.Client, base: Path, path: Path, stop: threading.Event) -> Result:
    if stop.is_set():                       # cancelled while queued behind others
        return Result(path, error="cancelled")
    try:
        with path.open("rb") as fh:
            r = client.put(f"/files/{path.relative_to(base).as_posix()}", content=fh)
        r.raise_for_status()
        return Result(path, status=r.status_code)
    except (httpx.HTTPError, OSError) as exc:
        return Result(path, error=str(exc) or type(exc).__name__)


def upload_all(client: httpx.Client, base: Path, files: list[Path], *,
               jobs: int, show_progress: bool) -> list[Result]:
    results: list[Result] = []
    stop = threading.Event()
    progress = Progress(TextColumn("uploading"), BarColumn(), MofNCompleteColumn(),
                        TextColumn("[red]{task.fields[failed]} failed"), TimeElapsedColumn(),
                        disable=not show_progress, transient=True)
    with progress, ThreadPoolExecutor(max_workers=jobs, thread_name_prefix="upload") as pool:
        task = progress.add_task("upload", total=len(files), failed=0)
        futures: dict[Future[Result], Path] = {
            pool.submit(upload_one, client, base, f, stop): f for f in files
        }
        failed = 0
        try:
            for future in as_completed(futures):
                result = future.result()     # upload_one never raises for item errors
                results.append(result)
                if result.error:
                    failed += 1
                    progress.console.print(f"[red]✗[/red] {result.path}: {result.error}")
                progress.update(task, advance=1, failed=failed)
        except KeyboardInterrupt:
            stop.set()
            pool.shutdown(wait=True, cancel_futures=True)
            raise
    return sorted(results, key=lambda r: r.path)

# src/mytool/cli.py
import sys
from pathlib import Path

import httpx
import typer

from mytool.upload import upload_all

app = typer.Typer()


@app.callback()
def main() -> None:
    """Artefact tools."""


@app.command()
def upload(
    directory: Path = typer.Argument(..., exists=True, file_okay=False),
    jobs: int = typer.Option(8, "--jobs", "-j", min=1, max=64, help="Uploads in parallel."),
    server: str = typer.Option("https://files.example.com", envvar="MYTOOL_SERVER"),
) -> None:
    """Upload every file under DIRECTORY."""
    files = sorted(p for p in directory.rglob("*") if p.is_file())
    limits = httpx.Limits(max_connections=jobs, max_keepalive_connections=jobs)
    with httpx.Client(base_url=server, limits=limits, timeout=httpx.Timeout(60.0, connect=5.0)) as client:
        try:
            results = upload_all(client, directory, files, jobs=jobs, show_progress=sys.stderr.isatty())
        except KeyboardInterrupt:
            typer.echo("interrupted — files already uploaded are kept", err=True)
            raise typer.Exit(130)
    failed = [r for r in results if r.error]
    typer.echo(f"uploaded {len(results) - len(failed)}/{len(files)} files", err=True)
    raise typer.Exit(1 if failed else 0)


if __name__ == "__main__":
    app()

The decisions behind it

Workers return values instead of raising. upload_one catches the expected failures (HTTP and file errors) and returns them in a Result. Unexpected exceptions — programming errors — still propagate through future.result() and crash loudly, which is what you want for bugs.

Printing happens on the main thread. Only the as_completed loop prints or updates the progress bar. Failures are printed through progress.console so they appear above the bar rather than tearing it. Workers never touch the terminal.

One shared client, sized to the pool. httpx.Client is thread-safe; sharing it shares the connection pool. httpx.Limits(max_connections=jobs) makes sure eight workers are not queueing for the default pool of connections, and that you never open more connections than workers.

Ctrl+C sets a flag and cancels the queue. Threads cannot be killed from outside, so cancellation is cooperative: cancel_futures=True drops everything not yet started, and the stop event lets queued-but-already-dequeued work bail out. Items in flight finish their current request. The command then exits 130, the convention for SIGINT.

Results are sorted. Completion order changes from run to run. Sorting before returning keeps any per-file output and JSON reports deterministic.

Choosing --jobs

For network-bound work, a default of 8 is conservative and safe for most services; 16–32 is reasonable for internal APIs you know can take it. Beyond that, the server's limits dominate — pair a larger pool with a rate limiter, as described in rate-limiting concurrent requests in CLIs. Always set a max: a user typing --jobs 1000 should get a validation error, not 1,000 threads.

UX considerations

• Progress by count, not by line. One bar with "37/400 · 2 failed" is more readable than 400 log lines. Print individual lines only for failures.
• Errors do not stop the batch. Users would rather see "398/400 uploaded, 2 failed" than an abort at the first problem. Make fail-fast an explicit --fail-fast option if some users need it.
• --jobs 1 is a debugging mode. Sequential execution makes output ordered and tracebacks easy to follow. Document it.
• Summarise at the end, on stderr. The count line and the exit code (non-zero if anything failed) are what scripts and people check.
• Watch out for thread-unsafe libraries. If an SDK client is not thread-safe, give each worker its own via ThreadPoolExecutor(initializer=...) and threading.local(), rather than sharing one.

Testing the behaviour

Test the batch function with a MockTransport server that can fail selectively, and assert on the set of results, the failure reporting and the ordering:

# tests/test_upload.py
from pathlib import Path

import httpx

from mytool.upload import upload_all


def make_files(tmp_path: Path, n: int) -> list[Path]:
    for i in range(n):
        (tmp_path / f"f{i:03}.txt").write_text(str(i))
    return sorted(tmp_path.iterdir())


def client_for(handler) -> httpx.Client:
    return httpx.Client(base_url="https://files.test", transport=httpx.MockTransport(handler))


def test_all_uploaded_and_sorted(tmp_path):
    files = make_files(tmp_path, 40)
    seen = []

    def handler(request):
        seen.append(request.url.path)
        return httpx.Response(201)

    results = upload_all(client_for(handler), tmp_path, files, jobs=8, show_progress=False)
    assert [r.path for r in results] == files
    assert all(r.status == 201 for r in results)
    assert len(seen) == 40


def test_failures_are_collected_not_raised(tmp_path):
    files = make_files(tmp_path, 10)

    def handler(request):
        return httpx.Response(500 if request.url.path.endswith("f007.txt") else 201)

    results = upload_all(client_for(handler), tmp_path, files, jobs=4, show_progress=False)
    failed = [r for r in results if r.error]
    assert [r.path.name for r in failed] == ["f007.txt"]
    assert len(results) == 10


def test_concurrency_is_bounded(tmp_path):
    import threading
    import time

    files = make_files(tmp_path, 30)
    active = 0
    peak = 0
    lock = threading.Lock()

    def handler(request):
        nonlocal active, peak
        with lock:
            active += 1
            peak = max(peak, active)
        time.sleep(0.01)
        with lock:
            active -= 1
        return httpx.Response(201)

    upload_all(client_for(handler), tmp_path, files, jobs=5, show_progress=False)
    assert 1 < peak <= 5

The bounded-concurrency test is the one that proves --jobs means what it says: it measures the peak number of simultaneous requests the fake server observed. MockTransport handlers run on the worker threads, which is why the counter needs a lock.

Conclusion

A thread pool is the lowest-effort, highest-return concurrency upgrade for an I/O-bound CLI: the worker code stays synchronous, and a bounded ThreadPoolExecutor with as_completed gives parallelism, smooth progress and per-item error handling. Keep printing on the main thread, share only thread-safe clients sized to the pool, cancel cooperatively on Ctrl+C, and sort results before output. When you outgrow it — hundreds of concurrent requests, or fine-grained cancellation — the same shape carries over to asyncio.

Frequently asked questions

Related

• Up: Concurrency and async in Python CLIs
• Multiprocessing for CPU-bound CLI tasks
• Rate-limiting concurrent requests in CLIs
• Adding progress bars and spinners to Python CLIs
• Downloading files with progress in Python