Input & UX

Reading Piped Input in Python CLIs

Read stdin safely in a Python CLI - detect piped data, support the dash convention, stream large inputs, and handle encodings without platform surprises.

Updated

A command that accepts piped input can be dropped into the middle of a pipeline, which is most of what makes a Unix tool useful. Getting it right is four small decisions: when to read stdin, how to let the user ask for it, how to stream it, and which encoding to assume.

TL;DR

  • Check sys.stdin.isatty() before reading — otherwise the tool hangs when run with no arguments.
  • Support - as a value meaning "read stdin", and document it in the help text.
  • Stream rather than read() for anything that might be large.
  • Name the encoding: sys.stdin.reconfigure(encoding="utf-8"), or use sys.stdin.buffer for bytes.
  • Empty input is a real case — produce a clear message, not a parser traceback.

Prerequisites

A CLI with a command that takes a document or a list of records. The examples use Typer; the sys.stdin handling is identical in Click and argparse.

Deciding whether to read stdin

Is there input waiting on stdin? A decision diagram: when standard input is not a terminal, data is being piped in; when it is, the tool should not block waiting for input that will never come. Is there input waiting on stdin? Is stdin a terminal? No — a pipe or a redirected file Read it data is waiting Yes — a person at a prompt Do not block require an argument instead Reading stdin unconditionally is why some tools appear to hang when run with no arguments.

sys.stdin.isatty() answers the practical question: is there a person at a keyboard, or is something feeding me data? When it returns False, stdin is a pipe or a redirected file, and reading it is safe. When it returns True, reading blocks forever on a user who has no idea they are supposed to type anything.

import sys
from pathlib import Path
from typing import Annotated

import typer

@app.command()
def apply(
    source: Annotated[Path, typer.Argument(help="Document to apply, or - for stdin.")] = Path("-"),
) -> None:
    text = read_document(source)
    ...

def read_document(source: Path) -> str:
    if str(source) != "-":
        return source.read_text(encoding="utf-8")

    if sys.stdin.isatty():
        raise typer.BadParameter("no input: pass a file, or pipe a document on stdin")

    text = sys.stdin.read()
    if not text.strip():
        typer.secho("no input on stdin (expected a JSON document)", fg="red", err=True)
        raise typer.Exit(65)

    return text

The empty-input branch is the one people leave out, and it is the case real users hit — an upstream command produced nothing, and without the check the failure is a JSONDecodeError traceback that says nothing about where the input was supposed to come from.

The dash convention

The "-" convention A table showing how a single dash is conventionally interpreted as standard input or standard output in command line arguments. The "-" convention Written as Means Example mytool apply - read the document from stdin cat job.json | mytool apply - mytool export -o - write results to stdout mytool export -o - | gzip mytool apply job.json read that file the ordinary case (no argument) your choice — document it usually an error or a default Supporting "-" costs a few lines and is what lets your tool appear anywhere in a pipeline.

A bare - where a filename is expected means "use the standard stream". It is a convention rather than a language feature, which means you implement it — and users of cat, jq, tar and every other Unix tool expect it to be there.

Click's File type handles it for you:

@click.command()
@click.argument("source", type=click.File("r"), default="-")
def apply(source) -> None:
    text = source.read()          # a file object, or stdin when the user typed -

In Typer, the equivalent is the explicit branch above, which is a few more lines and clearer about what happens. Either way, say so in the help text — "Document to apply, or - for stdin" — because a convention nobody documents is a convention half your users will not try.

Support it for output too. mytool export -o - writing to stdout lets someone pipe your results into gzip or another tool without a temporary file.

Streaming large inputs

sys.stdin.read() loads the whole stream into memory. For a config document that is fine; for a log file, an export or anything a user might generate, it is a memory limit you did not intend to set.

def read_records(source: Path) -> Iterator[dict]:
    """Yield one parsed record per line, from a file or stdin."""
    stream = sys.stdin if str(source) == "-" else source.open(encoding="utf-8")
    try:
        for number, line in enumerate(stream, start=1):
            line = line.strip()
            if not line:
                continue
            try:
                yield json.loads(line)
            except json.JSONDecodeError as exc:
                raise InputDataError(f"line {number}: {exc.msg}") from exc
    finally:
        if stream is not sys.stdin:
            stream.close()

Three properties are worth copying. It yields rather than collecting, so memory stays flat whatever the input size. It reports the line number in errors, which is the difference between a fixable message and a shrug on a million-line file. And it closes the file it opened while leaving stdin alone, because closing stdin in a library function is the kind of surprise that takes an afternoon to find.

Encodings and binary data

Text, bytes and encodings at the boundary Practices for handling encodings on standard input and output in a command line tool, and the habits that produce mojibake or crashes. Text, bytes and encodings at the boundary Predictable Read and write UTF-8 explicitly, everywhere Use sys.stdin.buffer for binary data Set errors="replace" only where garbling beats crashing Write "\n" and let the stream translate it Breaks on somebody's machine Relying on the platform default encoding Assuming stdin is text when it may be a binary stream Writing "\r\n" by hand on Windows Decoding with the wrong codec and blaming the input The default encoding differs across platforms and locales; naming UTF-8 removes the variable.

sys.stdin decodes using the platform default, which varies by operating system and locale. On a machine with a non-UTF-8 locale, the same pipeline that works for you fails with a UnicodeDecodeError for someone else.

sys.stdin.reconfigure(encoding="utf-8", errors="strict")
sys.stdout.reconfigure(encoding="utf-8", newline="\n")

Two lines near the top of main() remove the variable entirely. Use errors="replace" only where garbled text genuinely beats a failure — for a log viewer, perhaps; not for a config parser, where silently mangling input is worse than refusing it.

When the input is not text at all — an image, a compressed archive, a binary protocol — read the buffer directly:

payload = sys.stdin.buffer.read()          # bytes, no decoding attempted

And when both are possible, decide from a flag rather than by guessing: a heuristic that inspects the first bytes is right most of the time and produces a baffling failure the rest.

UX considerations

Do not block silently. If your command is waiting for stdin, the user sees nothing. Either refuse (with a message naming the argument) or, if waiting is genuinely intended, say so on stderr: reading from stdin; press Ctrl-D when done.

Report progress on stderr. A command chewing through a large piped input should show that it is alive, and that narration must not enter the pipeline.

Handle Ctrl-C cleanly. Reading a long stream is exactly when people interrupt. Exit 130 without a traceback, as described in handling keyboard interrupt cleanly.

Never prompt while reading stdin. If your command consumes piped data, stdin is not available for questions — any confirmation must come from a flag. This is the constraint that most often reveals a command trying to be interactive and scriptable in the same code path.

Testing the behaviour

The runner supplies stdin directly, so every case is testable in-process:

def test_reads_a_document_from_stdin(cli):
    result = cli.invoke(app, ["apply", "-"], input='{"replicas": 3}')
    assert result.exit_code == 0

def test_empty_stdin_is_a_clear_error(cli):
    result = cli.invoke(app, ["apply", "-"], input="")
    assert result.exit_code == 65
    assert "no input" in result.stderr

def test_malformed_line_names_the_line_number(cli):
    result = cli.invoke(app, ["apply", "-"], input='{"ok": 1}\nnot json\n')
    assert result.exit_code == 65
    assert "line 2" in result.stderr

def test_file_argument_still_works(cli, tmp_path):
    doc = tmp_path / "job.json"
    doc.write_text('{"replicas": 3}')
    assert cli.invoke(app, ["apply", str(doc)]).exit_code == 0

Four tests covering stdin, the empty case, an error message that helps, and the file path — which between them are almost every way this code is used.

Conclusion

Reading stdin well is mostly about not surprising anyone: check before reading so the tool never hangs, honour - so it composes, stream so size is not a limit, and name the encoding so it behaves the same on every machine. Four small decisions, and your command can sit anywhere in a pipeline.

Frequently asked questions

Can I tell whether piped input is empty before reading it?

Not reliably — a pipe with no data yet looks the same as one that will never have any. Read and check what you got, and produce a message that names both the expectation and the source, so the user knows whether to look upstream or at their own command.

What if a command needs both piped data and a confirmation?

Take the confirmation as a flag. Once stdin carries data it cannot also carry answers, so a prompt would either consume part of your input or fail. This is a good example of a constraint that improves the design: the command becomes scriptable because it has to.

How do I read stdin in argparse?

Identically — sys.stdin is not framework-specific. argparse.FileType("r") also handles the - convention, at the cost of opening the file during parsing, which makes error handling slightly less controllable than opening it yourself.

Should I support reading multiple files as well as stdin?

If the command is a filter, yes: accept a list of paths, treat an empty list as "read stdin", and treat - in the list as stdin too. That matches cat, grep and everything else people expect to combine your tool with.

Is there a performance cost to iterating stdin line by line?

Negligible — Python buffers the underlying reads, so line iteration is not one syscall per line. The cost that does matter is what you do per line; parsing JSON a million times is the expensive part, not the reading.

Can I accept both a piped document and command-line overrides?

Yes, and the useful rule is that flags win. Read the document, apply it as the base, then overlay any values the user passed explicitly — the same precedence as configuration files. Document it in one line, because a user piping a document and also passing a flag has a reasonable expectation either way and only one of them is true.