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 usesys.stdin.bufferfor 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
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
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
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.