Async Scripts & CLIs in Python¶

Q: How do I run async code from a click or Typer command?

Wrap the async function with a ParamSpec-typed decorator that calls asyncio.run, placed under the framework's decorators and using functools.wraps. Decorating an async def directly made both click 8.5.0 and Typer 0.27.2 exit 0 without running the command.

Q: How do I read stdin in an asyncio script?

Read 64 KiB chunks with await asyncio.to_thread(sys.stdin.buffer.read1, 65536) and split lines on the loop. It processed a million lines in 0.03–0.04 s from pipes and files; a thread call per line managed only 38,000 lines per second.

Q: What exit code should an asyncio script return?

Return an integer from main and pass asyncio.run(main()) to sys.exit: 0 for success, a distinct code such as 2 for partial success, 75 for temporary failure, and 143 after handling SIGTERM. Never call sys.exit inside a task.

Q: How do I stop a cron job written with asyncio from overlapping itself?

Take a non-blocking fcntl.flock on a lock file before calling asyncio.run and exit 75 if it is held; give the job a deadline shorter than its interval. Pidfiles fail after SIGKILL.

Q: Do progress bars slow down asyncio?

One bar did not measurably: 0.144 s with a rich bar against 0.136 s without. Adding 200 rich bars to a live display took 2.66 s because each add_task re-rendered every row on the event loop thread.

Most asyncio writing assumes a long-running server, but a great deal of async Python runs as short-lived processes: an admin command that calls fifty APIs at once, a nightly sync job started by cron, a pipeline stage reading IDs from stdin. Scripts have their own failure modes, and most of them are silent. Measured on Python 3.14: an async def command under click 8.5.0 or Typer 0.27.2 exited with status 0 without running its body; reading stdin with asyncio.to_thread(readline) per line ran at 38,000 lines/s against 28 million for chunked reads; adding 200 rich progress bars to a live display stalled the event loop for 2.38 s; sys.exit() inside a TaskGroup child skipped a sibling task's cleanup; an unhandled SIGTERM killed the process with no cleanup at all; and a pidfile left behind by a SIGKILLed run blocked every later run, where an flock released itself. This section covers the patterns that make async scripts start fast, read input efficiently, report progress cheaply, end with an honest exit status and never run twice at once.

The parent section, Asyncio Fundamentals & Event Loop Architecture, covers the loop that each of these scripts creates and tears down; this topic covers the process around it.

Architectural principles¶

One asyncio.run per process invocation. Everything that owns sockets, tasks or futures is created inside it and closed before it returns.
The entry point decides the exit status. main returns an integer; only the top level calls sys.exit.
Do synchronous preconditions before starting the loop. Locks, argument validation and configuration checks are cheap and need no event loop.
Bound everything. Concurrency, queue sizes, the visible progress rows and the total run time all have explicit limits.
Assume the process will be killed. Locks must release on death, and SIGTERM must lead to cleanup within a deadline.

Execution model: a loop that lives for one run¶

A script's event loop is created by asyncio.run, runs one coroutine, and is destroyed. Three consequences shape every pattern in this section. First, start-up cost is paid on every invocation: import asyncio took 26 ms of cumulative import time, click 15 ms, Typer 29 ms, rich's progress module 28 ms, httpx 81 ms and aiohttp 95 ms, and a click CLI answered --help in a median 52 ms against 115 ms for the same command under Typer. Second, objects bound to a loop do not survive it: an httpx.AsyncClient used under a second asyncio.run failed with RuntimeError: Event loop is closed, because its pooled connection belonged to the first. Third, the end of the loop is where cleanup happens — asyncio.run cancels and awaits any tasks still running — so anything that bypasses it, such as a SystemExit raised from a task or an unhandled signal, also bypasses cleanup.

The script's relationship with the outside world runs through three channels the event loop does not own: stdin, the exit status and signals. Each needs explicit handling. Stdin is a blocking file descriptor unless asyncio is told otherwise, and telling it otherwise (connect_read_pipe) fails for redirected files and changes the terminal's mode. The exit status is whatever sys.exit receives, or 1 for an uncaught exception, or a negative value for death by signal. Signals other than SIGINT are not handled by asyncio.run at all.

Pattern catalogue¶

Running async commands under click and Typer¶

Both frameworks call a decorated async def, get a coroutine back and drop it. A typed wrapper gives them a synchronous function instead:

def run_async[**P, R](fn: Callable[P, Coroutine[Any, Any, R]]) -> Callable[P, R]:
    @functools.wraps(fn)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        return asyncio.run(fn(*args, **kwargs))
    return wrapper


@click.command()
@click.option("--n", default=3)
@run_async
async def fetch(n: int) -> None:
    ...

functools.wraps is required for Typer, which reads the signature through __wrapped__; without it --n 5 was rejected and the command exited 2. asyncclick runs async commands and group callbacks on one loop natively. See writing async CLI commands with click and Typer.

Progress for concurrent work¶

One bar for the whole job costs almost nothing; many rich bars added to a live display cost a great deal, because each add_task re-renders every row on the event loop thread:

with Progress(refresh_per_second=10) as progress:
    overall = progress.add_task("downloading", total=len(urls))
    await asyncio.gather(*(fetch(u, lambda: progress.update(overall, advance=1)) for u in urls))

Measured over 200 tasks × 50 updates: 0.144 s with one rich bar, 0.135 s with one tqdm bar, 0.136 s with none — and 2.66 s with 200 per-task rich bars added while live. See showing progress for concurrent tasks.

Reading stdin without a thread hop per line¶

Read 64 KiB chunks in a thread and split them on the loop:

async def read_lines(chunk_size: int = 1 << 16):
    rest = b""
    while chunk := await asyncio.to_thread(sys.stdin.buffer.read1, chunk_size):
        lines = (rest + chunk).split(b"\n")
        rest = lines.pop()
        for line in lines:
            yield line
    if rest:
        yield rest

It handled a million lines in 0.03–0.04 s from pipes and redirected files alike; connect_read_pipe managed 1.69 million lines/s from a pipe but raised ValueError for < file. See reading stdin asynchronously.

Exit statuses that mean something¶

main returns a status, child tasks raise ordinary exceptions, and the entry point maps exceptions to statuses after the loop has cleaned up:

async def main() -> int:
    ...
    return EX_PARTIAL if failed else EX_OK

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

A SIGTERM handler that cancels the main task turns a kill with status -15 and no cleanup into status 143 with cleanup. See exiting async scripts with the right status code.

Never running twice at once¶

A non-blocking flock, taken before the loop starts, makes an overlapping run exit 75 in milliseconds and is released by the kernel even after SIGKILL:

fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o644)
try:
    fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
    sys.exit(75)

Waiting for the lock must poll with LOCK_NB; a blocking flock in to_thread acquired the lock after its timeout had fired. See preventing overlapping runs of async cron scripts.

Choosing how a script runs its async code¶

Not every script needs the same structure, and the cheapest correct one is usually best. A one-shot script with a single async phase needs nothing more than sys.exit(asyncio.run(main())). A CLI with several commands, a few of them async, wraps those commands individually and keeps the rest synchronous, so --help and argument errors never import the async client libraries. A tool whose commands share an async client or pool — a database admin CLI, say — benefits from asyncclick, where group and subcommand callbacks run on one loop and resources registered with ctx.with_async_resource are closed when the invocation ends. A script that alternates between async phases and synchronous work it cannot move into the loop, such as a notebook-style migration, reuses one loop explicitly with asyncio.Runner, as in reusing one loop across calls.

Whatever the structure, the scheduling layer around it decides the remaining guarantees. Under cron, the script must lock itself; under a systemd timer or a Kubernetes CronJob with concurrencyPolicy: Forbid, the scheduler already prevents overlap and the lock is a second line of defence. Under any orchestrator, SIGTERM is the normal way a run ends early, so the handler and the cleanup deadline are part of the design, not an afterthought.

Resource boundaries¶

Scripts fail by exhausting something as often as servers do, just faster. The limits that matter:

Concurrency. A fixed number of worker tasks — 32 in the example below — bounds sockets, file descriptors and upstream load. Pair it with a rate limit when the upstream has a quota, as in Rate Limiting & Throttling.
Queue size. A bounded queue between the stdin reader and the workers is the backpressure path: with yes as an endless producer, a 128-item queue held peak RSS at 20.2 MiB while 27,500 lines per second went through.
Visible progress rows. At most the concurrency limit plus one overall bar; per-row cost grows with the number of rows.
Total run time. An asyncio.timeout shorter than the schedule interval, so a hung run cannot hold its lock until someone notices.
Output. Results written to stdout pass through a buffer, and a downstream reader that exits early — python enrich.py | head — makes the next flush raise BrokenPipeError. Catch it at the entry point and exit quietly with the status you would otherwise have returned, rather than printing a traceback.
Cleanup time. Its own short deadline inside finally, under the orchestrator's grace period — 30 s by default in Kubernetes, 90 s in systemd.

Integrated production example¶

enrich.py reads IDs from stdin, looks each one up concurrently, writes JSON lines to stdout and reports partial failure through its exit status. It combines every pattern above in under a hundred lines:

"""enrich.py - read IDs from stdin, look each up concurrently, write JSON lines to stdout."""
import asyncio
import fcntl
import json
import os
import signal
import sys
from contextlib import contextmanager

import click

EX_OK, EX_PARTIAL, EX_TEMPFAIL = 0, 2, 75


@contextmanager
def single_instance(path: str):
    fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o644)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        os.close(fd)
        raise SystemExit(EX_TEMPFAIL)
    try:
        yield
    finally:
        os.close(fd)


async def read_lines(chunk_size: int = 1 << 16):
    rest = b""
    while chunk := await asyncio.to_thread(sys.stdin.buffer.read1, chunk_size):
        lines = (rest + chunk).split(b"\n")
        rest = lines.pop()
        for line in lines:
            if line.strip():
                yield line.decode().strip()
    if rest.strip():
        yield rest.decode().strip()


async def main(concurrency: int, deadline: float) -> int:
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGTERM, asyncio.current_task().cancel)
    queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=concurrency * 4)
    ok = failed = 0

    async def worker() -> None:
        nonlocal ok, failed
        while (item_id := await queue.get()) is not None:
            try:
                record = await lookup(item_id)          # your async client call
            except Exception as exc:
                failed += 1
                print(f"skip {item_id}: {exc}", file=sys.stderr)
            else:
                ok += 1
                sys.stdout.write(json.dumps(record) + "\n")

    try:
        async with asyncio.timeout(deadline):
            async with asyncio.TaskGroup() as tg:
                for _ in range(concurrency):
                    tg.create_task(worker())
                async for item_id in read_lines():
                    await queue.put(item_id)
                for _ in range(concurrency):
                    await queue.put(None)
    except asyncio.CancelledError:
        print(f"terminated after {ok} records", file=sys.stderr)
        return 128 + signal.SIGTERM
    except TimeoutError:
        print(f"deadline exceeded after {ok} records", file=sys.stderr)
        return EX_TEMPFAIL
    print(f"done: {ok} ok, {failed} failed", file=sys.stderr)
    return EX_PARTIAL if failed else EX_OK


@click.command()
@click.option("--concurrency", default=32, show_default=True)
@click.option("--deadline", default=240.0, show_default=True, help="seconds")
@click.option("--lock", default="/run/lock/enrich.lock", show_default=True)
def cli(concurrency: int, deadline: float, lock: str) -> None:
    with single_instance(lock):
        sys.exit(asyncio.run(main(concurrency, deadline)))

Run against a lookup that failed for IDs ending in 13 and otherwise took 2 ms, every exit path was exercised: seq 1 10000 | python enrich.py wrote 9,900 records and exited 2; the same input without the failing IDs exited 0; yes 1 | python enrich.py --deadline 1 exited 75 after 14,304 records; SIGTERM after 1.5 s exited 143 after 20,704 records, with the "terminated" line printed; and a second copy started while the first was running exited 75 immediately. The CLI layer is synchronous on purpose — click parses arguments and the lock is taken before any loop exists — and only main is async.

Diagnostic hook callout¶

The signals that tell you an async script is healthy are mostly outside the process:

Exit status distribution per job. Track counts of 0, 2, 75 and 143 per day. A rise in 2 means data or upstream problems; consecutive 75s mean runs are overlapping or a holder is stuck; 143 means the orchestrator is stopping runs before they finish.
Never-awaited warnings in stderr. Fail CI when stderr contains was never awaited; -W error::RuntimeWarning does not change the exit status of a script whose command coroutine was dropped.
Run duration against the deadline. Alert when the p95 run time passes 70% of the deadline; the next slow day will start producing 75s.
Loop lag during development. A lag probe alongside a progress display catches the add_task stall, a multi-second sample at the start of the run.

Thresholds: zero dropped-coroutine warnings, zero consecutive-75 streaks longer than two intervals, and cleanup completing within its budget on every 143.

Failure modes¶

Failure mode	Root cause	Detection	Fix
Command exits 0, does nothing	`async def` decorated directly by click or Typer	"was never awaited" in stderr	`run_async` wrapper or asyncclick
Stdin is the bottleneck	`to_thread(readline)` per line	~40k lines/s read rate	Chunked `read1` in a thread
`ValueError` on `< file`	`connect_read_pipe` on a regular file	Fails only when run with redirection	Chunked reader, or check stdin type
Loop stalls for seconds at start	Many rich bars added to a live display	Lag probe; slow `add_task`	Overall bar plus bounded rows
Sibling cleanup skipped	`sys.exit` inside a task	No cleanup log; traceback on stderr	Raise exceptions; status from `main`
No cleanup on stop	Unhandled `SIGTERM`	Status -15	Handler that cancels the main task
Every run refuses to start	Stale pidfile after `SIGKILL`	Consecutive 75s	`flock`, released by the kernel
Lock held after giving up	`to_thread(flock)` under a timeout	Next run sees lock held	Poll with `LOCK_NB`

Frequently Asked Questions¶

How do I run async code from a click or Typer command?