Concurrency and Async in Python CLIs

A CLI that processes one thing at a time is easy to write, easy to read and, for many jobs, far too slow. Checking the status of 300 repositories, uploading 2,000 files, fetching details for every item in a list, resizing a folder of images — each of these spends most of its time waiting on the network or burning a single CPU core while the other seven sit idle. Adding concurrency can turn a ten-minute command into a thirty-second one. It can also turn a predictable tool into one that hangs on Ctrl+C, floods an API until it is rate-limited, prints interleaved garbage, or loses errors somewhere inside a thread.

This topic covers adding concurrency to a Python CLI while keeping control of it: picking between threads, asyncio and processes based on the work; keeping all concurrency behind one function so commands stay simple; bounding how much runs at once; cancelling cleanly when the user presses Ctrl+C; and staying within an API's rate limits. It is part of the CLI Runtime & Systems Integration section and pairs naturally with calling HTTP APIs and running subprocesses, which supply most of the work worth parallelising.

TL;DR

• Match the model to the work. Waiting on network or disk: a ThreadPoolExecutor or asyncio. Computing in pure Python: a ProcessPoolExecutor.
• Keep concurrency inside one function that takes inputs and returns results. Commands stay synchronous and testable.
• Always bound concurrency with a pool size or semaphore, and expose it as --jobs.
• Collect failures as values, report them together, and pick the exit code from the whole batch.
• Make Ctrl+C stop everything: cancel pending work, let running work clean up, and exit 130.
• Respect rate limits with a token bucket when calling APIs; more workers do not beat a quota.

Choosing a concurrency model

Python gives you three practical models, and the choice follows from one question: while a unit of work is running, is it waiting or computing?

Threads (concurrent.futures.ThreadPoolExecutor) are the most direct upgrade for an existing synchronous CLI. Every blocking library — httpx.Client, boto3, subprocess, file I/O — releases the global interpreter lock (GIL) while it waits, so eight threads can wait on eight network requests at once. You keep your existing sync code and wrap the loop.

asyncio runs many tasks on one thread, switching between them whenever one awaits. It scales to thousands of concurrent network operations with little overhead and has first-class cancellation, but it requires async-capable libraries (httpx.AsyncClient, asyncpg, aiofiles) and async functions all the way down the call chain.

Processes (concurrent.futures.ProcessPoolExecutor) are the answer when the work is CPU-bound Python — parsing, hashing, image manipulation in pure Python, data transformation. Each worker is a separate interpreter with its own GIL, so work runs truly in parallel across cores, at the cost of startup time and copying arguments and results between processes.

A rough decision rule that holds up well in practice: start with a thread pool; move to asyncio when you need hundreds of concurrent operations or fine-grained cancellation; use processes only when profiling shows the CPU is the bottleneck.

How much faster, really?

For latency-bound work the gains are dramatic and easy to predict. If each of 40 requests takes 250 ms of mostly waiting, running them one after another takes ten seconds; with eight in flight at once, the ideal is a little over one second.

The curve flattens for three reasons, and each one is a design constraint rather than a bug: the server has its own limits (and will start returning 429s), connection setup adds cost per worker, and your local resources — file descriptors, bandwidth, memory — are finite. That is why every pattern in this topic includes an explicit bound.

Keep concurrency behind one function

The single most useful structural rule: commands should not know that anything runs concurrently. A command parses arguments, calls one function that owns the pool or event loop, receives ordinary results, and renders them. Threads, tasks and futures never escape that function.

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

from collections.abc import Callable, Iterable
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from typing import Generic, TypeVar

T = TypeVar("T")
R = TypeVar("R")


@dataclass(frozen=True)
class Outcome(Generic[T, R]):
    item: T
    result: R | None = None
    error: BaseException | None = None

    @property
    def ok(self) -> bool:
        return self.error is None


def run_all(fn: Callable[[T], R], items: Iterable[T], *, jobs: int = 8,
            on_done: Callable[[Outcome[T, R]], None] | None = None) -> list[Outcome[T, R]]:
    """Run fn over items with at most `jobs` in flight; never raises for item failures."""
    outcomes: list[Outcome[T, R]] = []
    with ThreadPoolExecutor(max_workers=jobs) as pool:
        futures = {pool.submit(fn, item): item for item in items}
        try:
            for future in as_completed(futures):
                item = futures[future]
                try:
                    outcome = Outcome(item, result=future.result())
                except Exception as exc:          # one bad item must not sink the batch
                    outcome = Outcome(item, error=exc)
                outcomes.append(outcome)
                if on_done:
                    on_done(outcome)
        except KeyboardInterrupt:
            pool.shutdown(wait=True, cancel_futures=True)
            raise
    return outcomes

The command using it is ordinary synchronous Typer code:

# src/mytool/cli.py
import httpx
import typer

from mytool.batch import run_all

app = typer.Typer()


@app.callback()
def main() -> None:
    """Link checker."""


@app.command()
def check(urls_file: typer.FileText, jobs: int = typer.Option(8, "--jobs", "-j", min=1, max=64)) -> None:
    """Check that every URL in URLS_FILE responds."""
    urls = [line.strip() for line in urls_file if line.strip()]
    with httpx.Client(timeout=10.0, follow_redirects=True) as client:
        def probe(url: str) -> int:
            return client.head(url).status_code

        outcomes = run_all(probe, urls, jobs=jobs)
    failed = [o for o in outcomes if not o.ok or o.result >= 400]
    for o in failed:
        typer.echo(f"✗ {o.item}: {o.error or o.result}", err=True)
    typer.echo(f"{len(urls) - len(failed)}/{len(urls)} ok", err=True)
    raise typer.Exit(1 if failed else 0)


if __name__ == "__main__":
    app()

Three habits are embedded here. Failures are values: each item's exception is captured in its Outcome, so one bad URL does not abort the other 299 and the command can report every failure at the end. The pool is bounded by --jobs, with a sensible default and a maximum. One httpx.Client is shared across threads — httpx clients are thread-safe and share a connection pool, which is exactly what you want. Parallelising CLI work with thread pools develops this into a full pattern with a Rich progress bar.

Async commands

Typer and Click call your command functions synchronously, so an async def command does not run on its own. The bridge is asyncio.run(), called once at the edge:

import asyncio

import httpx
import typer

app = typer.Typer()


async def fetch_all(urls: list[str], limit: int) -> list[int]:
    sem = asyncio.Semaphore(limit)
    async with httpx.AsyncClient(timeout=10.0) as client:
        async def one(url: str) -> int:
            async with sem:
                return (await client.get(url)).status_code

        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(one(u)) for u in urls]
    return [t.result() for t in tasks]


@app.command()
def statuses(urls: list[str], jobs: int = 16) -> None:
    """Print the status code of each URL."""
    for url, code in zip(urls, asyncio.run(fetch_all(urls, jobs))):
        typer.echo(f"{code}  {url}")

asyncio.TaskGroup (Python 3.11+) is the structured way to run tasks: if one fails, the others are cancelled, and no task can outlive the async with block. The semaphore bounds concurrency exactly as a pool size does. Running async code in Typer and Click covers a reusable decorator, async callbacks, and mixing sync and async libraries.

CPU-bound work and the GIL

Threads do not speed up pure-Python computation in standard CPython builds, because only one thread executes Python bytecode at a time. For CPU-bound work, use a process pool. The API is nearly identical — ProcessPoolExecutor instead of ThreadPoolExecutor — but the constraints differ: functions and arguments must be picklable, worker functions must be defined at module level, and the CLI's entry point must be guarded so child processes do not re-run it. Free-threaded Python builds (3.13t and later) remove the GIL, and are worth watching, but for tools distributed to other people today, processes remain the portable answer. Multiprocessing for CPU-bound CLI tasks covers the start-method differences between Linux, macOS and Windows, and chunking work so pickling costs do not eat the gains.

Stopping: Ctrl+C and failures

Starting concurrent work is easy; stopping it is where CLIs misbehave. When the user presses Ctrl+C they expect the tool to stop now — not after the remaining 280 queued items, and not by leaving half-written files and orphaned threads. The requirements are the same across all three models:

1. Stop scheduling new work immediately.
2. Cancel queued work that has not started.
3. Let running work finish or clean up, within a short bound.
4. Report what was done, what was cancelled, and what never started.
5. Exit with 130, the conventional code for termination by SIGINT.

For thread pools, shutdown(cancel_futures=True) handles steps 1 and 2; running threads cannot be killed and must check a flag or finish their current item. For asyncio, asyncio.run (since 3.11) converts the first Ctrl+C into cancellation of the main task, which propagates to every child in a TaskGroup. The details — shielding critical writes, bounding cleanup time, handling the second Ctrl+C — are in cancelling async tasks on Ctrl+C, building on the synchronous basics in handling KeyboardInterrupt cleanly.

Output from concurrent work

Concurrent work produces output concurrently, and a terminal is a single shared resource. Three rules keep it readable:

• Print from one place. Let workers return results and let the thread that collects them do the printing. The on_done callback above runs on the main thread, which is why it is safe to print or update a progress bar there.
• Prefer a progress bar to a log of every item. For hundreds of items, a single Rich progress bar with a count and an error tally is far more informative than hundreds of lines. Print individual lines only for failures.
• Keep results in a deterministic order when they go to stdout. as_completed yields in completion order, which differs every run; sort before printing so output is diffable, as described in emitting JSON output for scripting.

Shared state between workers

The fastest way to introduce a bug that appears once a week is to let workers mutate shared state. A dictionary of results updated from eight threads, a counter incremented without a lock, a list appended to from callbacks on different threads — each works in testing and fails under load, because individual Python operations are atomic but sequences of them are not. counts[key] = counts.get(key, 0) + 1 is a read, a compute and a write, and two threads can interleave between them.

The patterns that avoid it, in order of preference:

• Return, do not mutate. Workers compute a value and return it; the collecting loop — on a single thread — builds whatever structure it needs. The Outcome list above is built this way, and it needs no locks at all.
• Pass messages. When workers must report progress while running, put events onto a queue.Queue (or an asyncio.Queue) and drain it from one place.
• Lock narrowly. If shared mutable state is unavoidable — a cache that workers read and fill — protect it with a threading.Lock held only for the few lines that touch it, never around I/O.
• Share only thread-safe objects. HTTP clients, loggers and Rich consoles are designed to be shared. Database connections, most SDK clients created with mutable session state, and open files generally are not; create one per worker with threading.local() or an executor initializer.

In asyncio the risk is smaller — tasks only switch at await points — but not zero: any await between reading and writing shared state is a place another task can change it. The same "return, do not mutate" rule removes the question entirely.

Being a good API citizen

Concurrency multiplies your request rate, and most APIs enforce limits: a number of concurrent requests, a number per second, or both. A semaphore caps the first; a token bucket caps the second. Without a rate limiter, a fast CLI with sixteen workers can exhaust a quota in seconds and spend the rest of its run retrying 429s. Rate-limiting concurrent requests in CLIs implements a token bucket for threads and for asyncio and shows how to derive sensible settings from an API's documented limits.

Testing concurrent code

Concurrency adds non-determinism, and tests must not depend on timing luck. Three techniques keep them reliable:

• Test the batch function with trivial work. run_all(lambda x: x * 2, range(100), jobs=8) exercises the pool without I/O; assert on the set of results, not their order.
• Inject failures deliberately. A worker that raises for item 13 proves that other items still complete and that the failure is reported.
• Replace time. Rate limiters and retry loops take a clock and a sleep function as parameters, so tests advance a fake clock instead of waiting.

from mytool.batch import run_all


def test_failures_do_not_stop_the_batch():
    def work(n: int) -> int:
        if n == 13:
            raise ValueError("unlucky")
        return n * 2

    outcomes = run_all(work, range(50), jobs=8)
    assert len(outcomes) == 50
    assert sorted(o.result for o in outcomes if o.ok) == sorted(n * 2 for n in range(50) if n != 13)
    [bad] = [o for o in outcomes if not o.ok]
    assert bad.item == 13 and isinstance(bad.error, ValueError)

Key takeaways

• Pick threads or asyncio for waiting, processes for computing; start with a thread pool.
• Put all concurrency inside one function that returns plain results; commands stay synchronous.
• Bound concurrency, expose it as --jobs, and cap it at something sensible.
• Capture per-item failures as values and report them together at the end.
• Design for Ctrl+C from the start: cancel pending work, clean up running work, report, exit 130.
• Print from one thread, prefer progress bars, and sort anything that goes to stdout.
• Add a rate limiter before an API adds one for you.

Frequently asked questions

Related

• Up: CLI Runtime & Systems Integration
• Down: Running async code in Typer and Click
• Down: Parallelising CLI work with thread pools
• Down: Multiprocessing for CPU-bound CLI tasks
• Down: Cancelling async tasks on Ctrl+C
• Down: Rate-limiting concurrent requests in CLIs
• Sideways: Calling HTTP APIs from Python CLIs
• Sideways: CLI startup performance and lazy loading