Streaming Subprocess Output in Real Time

Your CLI runs a build, a test suite or a migration that takes minutes, and with subprocess.run(capture_output=True) the user stares at a silent terminal until it finishes — then gets the entire log at once. They want to see each line as the child produces it, ideally prefixed with which step it came from, while your tool still keeps a copy to inspect afterwards. This guide shows how to stream output from a child process line by line with Popen, why output sometimes still arrives in bursts even when you read it correctly, and how to handle stdout and stderr separately without deadlocking. It builds on the safe-call helper from calling external commands safely with subprocess.

Prerequisites

• Python 3.10+ and a Typer or Click CLI.
• Familiarity with subprocess.run() and its keyword arguments.
• A long-running command to try it on. The examples use a tiny Python script as the child so you can reproduce everything without extra tools.

Here is that child. Save it as slow.py; it prints a line every half second, with a warning on stderr in the middle:

# slow.py — a stand-in for a build or test run
import sys
import time

for i in range(1, 7):
    print(f"step {i}/6 done")
    if i == 3:
        print("warning: cache miss, rebuilding", file=sys.stderr)
    time.sleep(0.5)

The recipe: iterate over the pipe

Popen starts the child and returns immediately. With stdout=subprocess.PIPE and text=True, proc.stdout is a file object you can iterate; each iteration blocks until the child has written a complete line, then hands it to you. Nothing is accumulated unless you choose to keep it.

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

import subprocess
import sys
from collections import deque
from collections.abc import Callable


def stream_command(
    argv: list[str],
    *,
    on_line: Callable[[str], None],
    keep_last: int = 50,
    env: dict[str, str] | None = None,
) -> tuple[int, list[str]]:
    """Run argv, call on_line for each output line, return (code, tail)."""
    tail: deque[str] = deque(maxlen=keep_last)
    with subprocess.Popen(
        argv,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,      # one ordered stream
        stdin=subprocess.DEVNULL,
        text=True,
        encoding="utf-8",
        errors="replace",
        bufsize=1,                     # line-buffered on our side
        env=env,
    ) as proc:
        assert proc.stdout is not None
        for raw in proc.stdout:
            line = raw.rstrip("\n")
            tail.append(line)
            on_line(line)
    return proc.returncode, list(tail)

And the command that uses it:

# src/mytool/cli.py
import os
import sys

import typer

from mytool.stream import stream_command

app = typer.Typer()


@app.callback()
def main() -> None:
    """Build helpers."""


@app.command()
def build(verbose: bool = typer.Option(False, "--verbose", "-v")) -> None:
    """Run the build and show its progress."""
    env = {**os.environ, "PYTHONUNBUFFERED": "1"}

    def show(line: str) -> None:
        if verbose or line.startswith(("step", "warning")):
            typer.echo(f"  build │ {line}", err=True)

    code, tail = stream_command([sys.executable, "slow.py"], on_line=show, env=env)
    if code != 0:
        typer.secho(f"build failed (exit {code}); last lines:", fg="red", err=True)
        for line in tail[-10:]:
            typer.echo(f"  {line}", err=True)
        raise typer.Exit(1)
    typer.secho("build finished", fg="green", err=True)


if __name__ == "__main__":
    app()

The deque(maxlen=...) is the important memory decision: you keep the last fifty lines for the failure report, not the whole log. A build that prints for an hour costs the same memory as one that prints for a second.

The with block matters too. When the loop ends because the child closed its stdout, leaving the block waits for the process and sets returncode. If an exception is raised inside the loop — a KeyboardInterrupt, say — the context manager still closes the pipe and waits, so no zombie is left behind. Pair it with the approach in handling KeyboardInterrupt cleanly if the child should be stopped too.

Why output still arrives in bursts

You run the code above against a real program and the lines appear twenty at a time. Your loop is not the problem. The child is buffering.

Most programs — Python, anything using C's stdio, many Go and Rust tools — check whether their stdout is a terminal. If it is, they flush after every line so a person sees output promptly. If it is a pipe, they switch to block buffering and flush only when a buffer of several kilobytes fills, because that is much faster for bulk data. Your CLI reads from a pipe, so it gets blocks.

The fix belongs on the child's side, and it depends on the child:

• Python children: set PYTHONUNBUFFERED=1 in the environment, or run with python -u. That is why the command above builds env with it.
• C programs using stdio: on Linux and macOS with GNU coreutils, prefix the command with stdbuf -oL to force line buffering.
• Tools with their own flag: many have one — grep --line-buffered, sed -u, jq --unbuffered, docker build --progress=plain.
• Programs that only behave on a terminal: run them under a pseudo-terminal with the standard-library pty module (POSIX only). The child then believes it is talking to a person, colour codes and all, which you may need to strip.

Reach for the pseudo-terminal last. It changes more than buffering — the child may start prompting, drawing progress bars with carriage returns, or emitting escape codes — and it does not exist on Windows.

Two streams without a deadlock

Merging stderr into stdout (stderr=subprocess.STDOUT) gives one ordered stream and is the right default for a live log. But sometimes you need them apart: you parse stdout as data and show stderr as diagnostics. The naive version reads one pipe, then the other:

# DON'T: can deadlock
out = proc.stdout.read()
err = proc.stderr.read()

If the child writes more to stderr than the pipe buffer holds (64 KB on Linux) while you are blocked reading stdout, the child blocks on its write, you block on your read, and neither moves. The fix is to read both concurrently. Threads are the simplest portable way:

import subprocess
import threading
from collections.abc import Callable
from typing import IO


def _pump(stream: IO[str], handle: Callable[[str], None]) -> None:
    for line in stream:
        handle(line.rstrip("\n"))


def stream_two(argv: list[str], on_out: Callable[[str], None],
               on_err: Callable[[str], None]) -> int:
    with subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                          stdin=subprocess.DEVNULL, text=True, encoding="utf-8",
                          errors="replace") as proc:
        threads = [
            threading.Thread(target=_pump, args=(proc.stdout, on_out), daemon=True),
            threading.Thread(target=_pump, args=(proc.stderr, on_err), daemon=True),
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
    return proc.returncode

The callbacks run on the pumping threads, so keep them simple — appending to a list, putting onto a queue.Queue, or printing. print and typer.echo are safe to call from threads; building up a shared data structure needs a lock or a queue. If your CLI already uses asyncio, asyncio.create_subprocess_exec with two readline() tasks does the same job without threads; see running async code in Typer and Click.

UX considerations

• Prefix each line with its source. build │ ... makes it obvious which lines are yours and which are the child's, especially when a command runs several steps in sequence.
• Stream to stderr. Live progress is narration, not results. Keeping it on stderr lets mytool build > report.txt still capture only your command's actual output.
• Filter by default, pass everything with --verbose. Most child output is noise to your user. Show the lines that matter (steps, warnings, errors) and let -v reveal the rest.
• Keep a tail for failures. When the child fails, reprint its last lines under a clear heading. That is often the only part the user reads.
• Do not fight a progress display. If you use a Rich progress bar or status spinner, print child lines through its console (progress.console.print(...)) so they appear above the bar instead of tearing it. The details are in adding progress bars and spinners to Python CLIs.
• Strip carriage-return redraws. Tools that draw their own progress bars emit \r to overwrite a line. In a log those become enormous single lines; keep only the text after the last \r.

Testing the behaviour

Two properties are worth pinning: lines arrive before the child exits, and the returned tail and exit code are right. The first needs a real child that pauses between lines, so the test can observe the timing:

# tests/test_stream.py
import sys
import time

from mytool.stream import stream_command

CHILD = (
    "import sys, time\n"
    "for i in range(3):\n"
    "    print(i, flush=True)\n"
    "    time.sleep(0.3)\n"
    "sys.exit(2)\n"
)


def test_lines_arrive_while_child_runs():
    seen: list[tuple[str, float]] = []
    start = time.monotonic()
    code, tail = stream_command(
        [sys.executable, "-c", CHILD],
        on_line=lambda line: seen.append((line, time.monotonic() - start)),
    )
    assert code == 2
    assert [line for line, _ in seen] == ["0", "1", "2"]
    # The first line arrived well before the child finished (~0.9s).
    assert seen[0][1] < 0.6


def test_tail_is_bounded():
    code, tail = stream_command(
        [sys.executable, "-c", "for i in range(500): print(i)"],
        on_line=lambda _: None,
        keep_last=5,
    )
    assert code == 0
    assert tail == ["495", "496", "497", "498", "499"]

Timing assertions are inherently a little fragile on loaded CI machines, so leave a generous margin — the point is to catch a regression to "everything at the end", which shows up as the first line arriving after roughly a second. For the command layer, patch stream_command and feed it canned lines, as in testing Click commands with CliRunner.

Conclusion

Live output is three decisions: iterate over a Popen pipe rather than waiting on run(), make the child flush per line rather than per block, and read two streams concurrently or merge them into one. Get those right and a long-running child feels responsive, your memory use stays flat, and failures come with the context needed to fix them. When the child might run too long, add a deadline using handling subprocess timeouts and exit codes.

Frequently asked questions

Related

• Up: Running subprocesses from Python CLIs
• Calling external commands safely with subprocess
• Handling subprocess timeouts and exit codes
• Detecting a TTY and adapting output
• Adding progress bars and spinners to Python CLIs