Python 基础体系 · 第 50/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python CLI 工程：argparse、子命令、退出码、管道和可测试性

命令行工具（Command-Line Interface，CLI）是一个由终端、参数解析器、业务逻辑、标准输入输出、子进程和操作系统退出状态共同组成的接口。它不是“给函数加几个参数”这么简单：调用者不仅关心程序做了什么，也关心命令格式是否稳定、输出写到哪里、失败时返回什么退出码、能否接入管道，以及测试时是否必须真的启动一个进程。

本文以 Python 3.14 标准库为范围，从一个可执行的多子命令工具开始，逐步解释 argparse、子命令、退出码、Unix 管道、subprocess 和可测试性之间的关系。

1. CLI 的真实边界：参数、数据流和进程状态

一个 CLI 程序至少有三种外部契约：

调用语法：命令名、位置参数、选项、子命令以及它们的组合方式。
数据流：标准输入 stdin、标准输出 stdout、标准错误 stderr 分别承载什么。
进程结果：程序结束时向操作系统报告的退出码。

可以把一次命令调用抽象为：

(\text{argv}, \text{stdin}, \text{环境}, \text{文件系统}) \rightarrow (\text{stdout}, \text{stderr}, \text{exit code})

其中：

argv 是命令行参数序列，例如 ["count", "--unique", "a.txt"]；
stdin 是可选的输入字节流；
环境包括当前目录、环境变量、权限和可执行文件搜索路径；
文件系统是输入文件、输出文件和工作目录；
stdout 应承载机器或下游程序需要的数据；
stderr 应承载诊断信息、进度和警告；
exit code 是调用者判断成功或失败的主要信号。

这三个契约必须分开设计。下面的输出虽然“看起来方便”，但会破坏管道：

print("reading input...", file=sys.stdout)
print("42")

如果调用者执行：

tool count input.txt | another-tool

那么 another-tool 会同时收到诊断信息和真正的数据。更可靠的约定是：

print("reading input...", file=sys.stderr)
print("42", file=sys.stdout)

这不是 Python 特有的语法，而是 Unix 命令行生态中长期形成的流式组合模型。stdout 是数据，stderr 是旁路诊断，退出码是状态。

2. `argparse` 的职责：把文本参数转换为结构化配置

argparse 是 Python 标准库中的命令行解析模块。它从参数序列中识别位置参数、选项和值，并将结果放入 argparse.Namespace；同时，它能够生成帮助信息，并在参数非法时报告错误。(docs.python.org)

一个最小示例：

# simple_cli.py
from __future__ import annotations

import argparse


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="simple-cli",
        description="统计文本文件中的非空行数。",
    )
    parser.add_argument(
        "path",
        help="输入文件路径；使用 - 表示从标准输入读取",
    )
    parser.add_argument(
        "-u",
        "--unique",
        action="store_true",
        help="只统计去重后的行",
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)

    print(args.path)
    print(f"unique={args.unique}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

运行：

python simple_cli.py data.txt

预期输出：

data.txt
unique=False

这里有三个重要边界。

2.1 `parse_args()` 不是业务逻辑

parse_args() 只应负责：

字符串参数
    ↓
语法检查
    ↓
类型转换
    ↓
Namespace

它不应负责：

打开大量文件；
发起网络请求；
修改数据库；
执行外部命令；
打印业务结果；
决定某个业务错误对应哪个退出码。

例如，不推荐这样写：

parser.add_argument("input", type=open)

argparse.FileType 确实可以在解析期间打开文件，并支持用 - 表示标准输入或标准输出；但是，如果后续参数解析失败，已经打开的文件可能不会自动关闭，输出文件也可能提前被截断。官方文档因此建议在参数解析完成后再用 with 管理文件。(docs.python.org)

更稳妥的写法是：

parser.add_argument("input")

然后在业务层打开：

from pathlib import Path
import sys
from typing import TextIO


def open_input(path: str) -> TextIO:
    if path == "-":
        return sys.stdin
    return Path(path).open("r", encoding="utf-8")

调用者可以明确控制资源生命周期：

def read_lines(path: str) -> list[str]:
    if path == "-":
        return list(sys.stdin)

    with open(path, "r", encoding="utf-8") as stream:
        return list(stream)

2.2 `type` 是转换，不是完整验证

type=int 只保证输入能转换成整数：

parser.add_argument("--workers", type=int)

它接受：

tool --workers 4

但也接受：

tool --workers -1

如果业务要求大于零，应将约束写成可复用的转换函数：

def positive_int(text: str) -> int:
    value = int(text)
    if value <= 0:
        raise argparse.ArgumentTypeError(
            f"必须是正整数，得到：{text!r}"
        )
    return value

然后：

parser.add_argument("--workers", type=positive_int, default=1)

此时失败属于命令行参数错误，而不是业务执行错误。解析器会把 ArgumentTypeError 转换成用户可理解的命令行错误。

2.3 默认值必须区分“未提供”和“显式提供”

下面两个状态有时不是一回事：

没有指定 --timeout
显式指定 --timeout 0

如果使用：

parser.add_argument("--timeout", type=float, default=0)

业务层只能看到 0，无法判断用户是否显式指定了它。

可以使用 argparse.SUPPRESS：

parser = argparse.ArgumentParser(
    argument_default=argparse.SUPPRESS
)
parser.add_argument("--timeout", type=float)

此时未提供参数时，返回的 Namespace 不会创建对应属性。官方文档说明，argument_default=argparse.SUPPRESS 可以抑制缺省参数属性的生成。(docs.python.org)

不过，这种设计会增加属性不存在的分支。更常见的工程方式是使用一个明确的哨兵值，或在解析后立即转换为业务配置对象：

from dataclasses import dataclass


@dataclass(frozen=True)
class Config:
    timeout: float | None


def to_config(args: argparse.Namespace) -> Config:
    return Config(timeout=getattr(args, "timeout", None))

3. 参数解析的三个层次：语法、类型和语义

CLI 参数验证可以分为三个层次。

第一层：语法合法

例如：

tool --verbose
tool --verbose=yes

如果 --verbose 是布尔开关，第二种写法通常不符合定义。

这属于 argparse 的职责。

第二层：类型合法

例如：

tool --workers hello

如果参数要求整数，字符串无法转换。

这通常由 type= 处理。

第三层：业务语义合法

例如：

tool copy source.txt destination.txt --mode append

语法和类型都合法，但如果目标文件已经存在，而 append 又不允许覆盖，那么这是业务规则冲突。

这应由业务层处理，而不是把所有逻辑塞进 type=：

def run_copy(args: argparse.Namespace) -> int:
    source = Path(args.source)
    destination = Path(args.destination)

    if destination.exists() and args.mode == "create":
        print(
            f"目标已存在：{destination}",
            file=sys.stderr,
        )
        return 3

    destination.write_bytes(source.read_bytes())
    return 0

这样，参数解析与业务执行的错误路径是可区分的：

解析失败       → argparse 输出错误 → 通常退出码 2
业务规则失败   → 业务函数返回退出码
外部命令失败   → 捕获 CalledProcessError 或检查 returncode
程序内部异常   → 异常传播或统一转换

Unix 程序通常使用退出码 2 表示命令行语法错误，1 表示其他一般错误，但这属于约定而不是 Python 对所有应用的强制规范。sys.exit() 对整数参数的基本语义是：0 表示成功，非零表示异常终止。(docs.python.org)

4. 子命令：把一个程序拆成多个独立语法空间

当一个工具同时提供多种操作时，可以使用子命令：

wr count FILE
wr filter PATTERN FILE
wr run COMMAND ...

每个子命令拥有独立的参数空间：

wr count
    --unique
    FILE

wr filter
    --ignore-case
    PATTERN
    FILE

wr run
    --timeout SECONDS
    COMMAND ...

argparse.add_subparsers() 用于创建子命令集合，随后通过 add_parser() 为每个子命令创建解析器。子解析器只会向命名空间加入当前选中的命令及其参数，不会加入其他子命令的参数。(docs.python.org)

最关键的调用方式是把处理函数放进子解析器的默认值：

subparser.set_defaults(func=handler)

解析完成后直接执行：

args.func(args)

官方文档将这种模式作为处理子命令的有效方式。(docs.python.org)

5. 一个完整的多子命令工具

下面实现一个名为 wr 的工具：

count：统计非空行；
filter：筛选包含指定文本的行；
run：执行外部命令并传递标准输入；
所有业务处理函数返回整数退出码；
顶层 main() 负责统一处理异常和退出。

# wr_cli.py
from __future__ import annotations

import argparse
import subprocess
import sys
from pathlib import Path
from typing import Sequence, TextIO


EX_USAGE = 64
EX_DATAERR = 65
EX_NOINPUT = 66
EX_UNAVAILABLE = 69
EX_SOFTWARE = 70


def positive_int(text: str) -> int:
    value = int(text)
    if value <= 0:
        raise argparse.ArgumentTypeError(
            f"必须是正整数，得到：{text!r}"
        )
    return value


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="wr",
        description="面向管道的文本与命令行工具。",
        allow_abbrev=False,
        suggest_on_error=True,
        color=False,
    )

    parser.add_argument(
        "--version",
        action="version",
        version="wr 1.0",
    )

    subparsers = parser.add_subparsers(
        dest="command",
        required=True,
        title="子命令",
    )

    count_parser = subparsers.add_parser(
        "count",
        help="统计非空行数",
        description="统计文件或标准输入中的非空行数。",
    )
    count_parser.add_argument(
        "path",
        help="输入路径；使用 - 表示标准输入",
    )
    count_parser.add_argument(
        "-u",
        "--unique",
        action="store_true",
        help="按去除行尾换行后的完整文本去重",
    )
    count_parser.set_defaults(func=run_count)

    filter_parser = subparsers.add_parser(
        "filter",
        help="筛选匹配的行",
        description="输出包含 PATTERN 的输入行。",
    )
    filter_parser.add_argument(
        "pattern",
        help="要查找的文本",
    )
    filter_parser.add_argument(
        "path",
        help="输入路径；使用 - 表示标准输入",
    )
    filter_parser.add_argument(
        "-i",
        "--ignore-case",
        action="store_true",
        help="忽略大小写",
    )
    filter_parser.set_defaults(func=run_filter)

    run_parser = subparsers.add_parser(
        "run",
        help="执行外部命令",
        description="执行 COMMAND，并将当前标准输入传给它。",
    )
    run_parser.add_argument(
        "--timeout",
        type=positive_int,
        default=None,
        help="超时时间，单位为秒",
    )
    run_parser.add_argument(
        "program",
        help="要执行的程序",
    )
    run_parser.add_argument(
        "args",
        nargs=argparse.REMAINDER,
        help="传递给程序的其余参数",
    )
    run_parser.set_defaults(func=run_command)

    return parser


def input_stream(path: str) -> tuple[TextIO, bool]:
    if path == "-":
        return sys.stdin, False

    try:
        return Path(path).open("r", encoding="utf-8"), True
    except OSError as exc:
        raise RuntimeError(f"无法打开输入文件 {path!r}: {exc}") from exc


def run_count(args: argparse.Namespace) -> int:
    stream, should_close = input_stream(args.path)

    try:
        if args.unique:
            seen: set[str] = set()
            for line in stream:
                text = line.rstrip("\r\n")
                if text.strip() and text not in seen:
                    seen.add(text)
            print(len(seen))
        else:
            total = sum(
                1
                for line in stream
                if line.strip()
            )
            print(total)
    finally:
        if should_close:
            stream.close()

    return 0


def run_filter(args: argparse.Namespace) -> int:
    stream, should_close = input_stream(args.path)
    needle = args.pattern.casefold() if args.ignore_case else args.pattern

    try:
        for line in stream:
            candidate = line.casefold() if args.ignore_case else line
            if needle in candidate:
                sys.stdout.write(line)
    finally:
        if should_close:
            stream.close()

    return 0


def run_command(args: argparse.Namespace) -> int:
    command = [args.program, *args.args]

    try:
        completed = subprocess.run(
            command,
            stdin=sys.stdin,
            stdout=sys.stdout,
            stderr=sys.stderr,
            timeout=args.timeout,
            check=False,
        )
    except FileNotFoundError:
        print(
            f"找不到可执行程序：{args.program!r}",
            file=sys.stderr,
        )
        return EX_UNAVAILABLE
    except subprocess.TimeoutExpired:
        print(
            f"命令超时：{command!r}",
            file=sys.stderr,
        )
        return 124
    except OSError as exc:
        print(
            f"无法启动命令 {command!r}：{exc}",
            file=sys.stderr,
        )
        return EX_UNAVAILABLE

    if completed.returncode < 0:
        # POSIX 下，负值表示被信号终止。
        print(
            f"命令被信号 {-completed.returncode} 终止",
            file=sys.stderr,
        )
        return 128 + (-completed.returncode)

    return completed.returncode


def main(
    argv: Sequence[str] | None = None,
) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)

    try:
        return args.func(args)
    except BrokenPipeError:
        # 下游提前关闭管道时，不再打印新的错误信息。
        return 0
    except RuntimeError as exc:
        print(f"错误：{exc}", file=sys.stderr)
        return EX_NOINPUT
    except KeyboardInterrupt:
        print("已取消", file=sys.stderr)
        return 130


if __name__ == "__main__":
    raise SystemExit(main())

这个程序可以直接运行：

printf 'apple\nbanana\napple\n\n' | python wr_cli.py count -

输出：

使用去重模式：

printf 'apple\nbanana\napple\n\n' | python wr_cli.py count --unique -

输出：

筛选并继续接管道：

printf 'INFO started\nERROR failed\nINFO done\n' \
  | python wr_cli.py filter ERROR - \
  | tr '[:lower:]' '[:upper:]'

输出：

ERROR FAILED

这里的数据流是：

flowchart LR
    A[终端或上游命令] -->|stdin| B[wr filter]
    B -->|匹配行 stdout| C[下游命令]
    B -->|错误与诊断 stderr| D[终端]
    B -->|整数退出码| E[操作系统]

关键点是：filter 的正常结果只写 stdout，错误写 stderr，下游命令不会把诊断文本误当作数据。

6. `required=True` 与没有子命令的状态

默认情况下，子命令不是必选的：

subparsers = parser.add_subparsers(dest="command")

此时：

wr

可能成功解析，得到：

Namespace(command=None)

如果程序没有处理 command is None，错误会在更后面以：

AttributeError: 'Namespace' object has no attribute 'func'

的形式暴露出来。这是错误分层失败：一个命令行语法错误被延迟成了内部异常。

更明确的方式是：

subparsers = parser.add_subparsers(
    dest="command",
    required=True,
)

于是缺少子命令会在解析阶段失败。required 参数用于指定是否必须提供子命令，Python 3.7 起支持该关键字参数。(docs.python.org)

如果希望无参数运行时显示帮助，可以显式处理：

def main(argv: Sequence[str] | None = None) -> int:
    parser = build_parser()

    if argv == []:
        parser.print_help()
        return 0

    args = parser.parse_args(argv)
    return args.func(args)

但这会引入一个语义选择：

wr          → 显示帮助并成功
wr          → 报错并返回 2

两者都可以成立，关键是保持稳定。对于自动化环境，通常“缺少必需子命令”更适合作为失败；对于交互式工具，显示顶层帮助可能更友好。

7. Python 3.14 的 `argparse` 能力与兼容性边界

Python 3.14 的 ArgumentParser 新增了两个与用户体验有关的参数：

parser = argparse.ArgumentParser(
    suggest_on_error=True,
    color=False,
)

suggest_on_error=True 可以对字符串类型的非法 choices 和错误的子命令名称给出建议；color 控制帮助输出中的颜色，默认值为 True。这些参数是在 Python 3.14 中加入的。(docs.python.org)

例如：

parser = argparse.ArgumentParser(
    suggest_on_error=True,
)
parser.add_argument(
    "--format",
    choices=["json", "text"],
)

执行：

python tool.py --format jso

可能得到类似：

argument --format: invalid choice: 'jso', maybe you meant 'json'?

但是，颜色输出不应被误认为“总能安全地用于日志文件”。帮助或错误信息可能包含 ANSI 转义序列；如果需要稳定的纯文本输出，应显式设置：

color=False

或者通过环境变量关闭颜色。官方文档还说明，重定向 stderr 时错误信息可能包含颜色控制码。(docs.python.org)

对于需要支持 Python 3.13 及更早版本的项目，直接传入 suggest_on_error=True 会导致构造函数不兼容。可以采用属性赋值的兼容写法：

parser = argparse.ArgumentParser()

if hasattr(parser, "suggest_on_error"):
    parser.suggest_on_error = True

但如果项目明确要求 Python 3.14，直接使用关键字参数更清楚。

另外，长选项默认允许无歧义缩写：

parser.add_argument("--verbose")

用户可能写：

tool --verb

只要当前不存在其他以 --verb 开头的选项，就可能被接受。CLI 一旦公开，后续新增 --verbose-output，原本可用的 --verb 就可能变成歧义。因此工具通常应设置：

allow_abbrev=False

这样参数名必须完整匹配。该参数用于控制长选项缩写，官方文档明确提供了关闭方式。(docs.python.org)

8. 退出码：进程如何向调用者报告结果

8.1 `return` 和 `sys.exit()` 的区别

业务函数推荐返回整数：

def run_count(args: argparse.Namespace) -> int:
    ...
    return 0

最外层再执行：

raise SystemExit(main())

sys.exit(code) 的本质是抛出 SystemExit 异常，而不是立即执行某种不可拦截的机器级终止。finally 清理逻辑仍会执行，外层也可以捕获这个异常。(docs.python.org)

因此：

def main() -> int:
    return 3


if __name__ == "__main__":
    raise SystemExit(main())

比在业务深处到处写：

sys.exit(3)

更容易测试。测试业务函数时直接检查返回值即可，不必捕获进程退出。

8.2 退出码的设计原则

一个 CLI 至少需要区分：

0   成功
2   参数或命令行语法错误
非零 业务失败、输入失败或外部命令失败

更细的退出码有助于脚本调用者采取不同动作：

EX_USAGE = 64       # 命令用法错误
EX_NOINPUT = 66     # 输入文件或输入资源不可用
EX_UNAVAILABLE = 69 # 依赖的外部程序不可用
EX_SOFTWARE = 70    # 程序内部错误

这些数值来自常见的 Unix sysexits 约定，不是 Python 解释器强制要求。项目可以选择自己的码表，但必须避免随意复用同一个值。

例如：

2  参数错误
3  业务规则不满足
4  输入数据格式错误
5  外部服务不可用
124 超时
130 用户通过 Ctrl-C 取消

重要的是建立稳定的映射：

\text{异常类别} \rightarrow \text{退出码}

而不是把异常文本当作机器接口。

8.3 `sys.exit("message")` 的陷阱

下面的写法会把消息输出到 stderr，并以退出码 1 结束：

sys.exit("input file not found")

这很方便，但不适合作为复杂 CLI 的统一错误机制，因为：

所有此类错误都倾向于变成退出码 1；
业务层无法清晰表达不同错误类型；
测试需要捕获 SystemExit；
错误格式容易分散到各处。

更适合的结构是：

class InputError(Exception):
    pass


def main(argv: list[str] | None = None) -> int:
    try:
        return execute(argv)
    except InputError as exc:
        print(f"输入错误：{exc}", file=sys.stderr)
        return 66

9. 外部命令：`subprocess.run()`、返回值和异常

Python 不会隐式通过 shell 解释普通的参数序列。推荐形式是：

completed = subprocess.run(
    ["git", "status", "--short"],
    capture_output=True,
    text=True,
    check=False,
)

subprocess.run() 会等待子进程结束并返回 CompletedProcess；capture_output=True 会捕获标准输出和标准错误，text=True 会以文本模式处理流。若 check=True 且子进程返回非零退出码，则抛出 CalledProcessError。(docs.python.org)

completed = subprocess.run(
    ["git", "status", "--short"],
    capture_output=True,
    text=True,
    check=False,
)

if completed.returncode != 0:
    print(completed.stderr, file=sys.stderr)
    return completed.returncode

print(completed.stdout, end="")
return 0

这段代码显式区分了三种故障：

程序不存在：
```
FileNotFoundError
```
程序启动了，但返回非零：
```
completed.returncode != 0
```
程序启动后超时：
```
subprocess.TimeoutExpired
```

如果使用 check=True：

try:
    completed = subprocess.run(
        ["git", "status"],
        capture_output=True,
        text=True,
        check=True,
    )
except subprocess.CalledProcessError as exc:
    print(
        f"git 失败，退出码={exc.returncode}",
        file=sys.stderr,
    )
    return 5

CalledProcessError 中包含命令参数、退出码以及在捕获时得到的输出。(docs.python.org)

9.1 不要把 shell 命令字符串当作参数序列

不安全或脆弱的写法：

filename = user_input
subprocess.run(f"cat {filename}", shell=True)

如果 filename 为：

data.txt; rm -rf /tmp/cache

那么 shell 可能将其解析为多条命令。

更安全的写法：

subprocess.run(
    ["cat", filename],
    check=True,
)

默认 shell=False 时，Python 不会自动调用系统 shell；参数中的 shell 元字符不会被解释。显式使用 shell=True 时，调用者必须负责正确处理空格、引号和元字符，并防止 shell 注入。(docs.python.org)

shell=True 并不是绝对禁止。它适用于确实需要 shell 语法的场景，例如：

subprocess.run(
    "printf '%s\\n' *.txt",
    shell=True,
    check=True,
)

但这段代码依赖 shell 的通配符展开，而且输入不能来自不可信来源。若只是执行一个程序，应优先使用列表参数。

10. 超时、信号和负退出码

外部命令至少有两个不同的时间概念：

启动时间
等待子进程完成的时间

subprocess.run(timeout=...) 的超时参数传递给底层 communicate()；超时后会终止子进程并等待，然后重新抛出 TimeoutExpired。进程创建本身在某些平台上无法被完整打断，因此超时异常不一定精确发生在指定秒数。(docs.python.org)

try:
    result = subprocess.run(
        ["slow-command"],
        timeout=10,
        check=False,
    )
except subprocess.TimeoutExpired:
    return 124

在 POSIX 系统中，CompletedProcess.returncode 为负值表示子进程被信号 N 终止：

if result.returncode == -signal.SIGTERM:
    ...

例如：

if result.returncode < 0:
    signal_number = -result.returncode
    return 128 + signal_number

128 + 信号编号 是 Unix shell 中常见的映射方式，但不同 shell、平台和调用层可能有差异。不要把它当作所有操作系统都提供的统一规范。

11. 管道的本质：前一个进程的 `stdout` 连接后一个进程的 `stdin`

Shell 命令：

producer | consumer

可以理解为：

producer.stdout ──管道──> consumer.stdin

在 Python 中，可以使用 Popen 显式建立同样的连接：

from subprocess import PIPE, Popen


producer = Popen(
    ["printf", "a\nb\nc\n"],
    stdout=PIPE,
    text=True,
)

consumer = Popen(
    ["grep", "b"],
    stdin=producer.stdout,
    stdout=PIPE,
    text=True,
)

assert producer.stdout is not None
producer.stdout.close()

output, _ = consumer.communicate()
producer.wait()

print(output, end="")

输出：

PIPE 表示为子进程创建新的管道；communicate() 会读写相关流、等待进程结束并设置返回码。官方文档特别提醒，直接对 stdout.read()、stderr.read() 或 stdin.write() 组合操作，可能因为操作系统管道缓冲区填满而死锁；使用 communicate() 可以避免这一类死锁。(docs.python.org)

producer.stdout.close() 也不是多余的。它关闭父进程持有的额外读端，使生产者在消费者提前退出时能够收到 SIGPIPE，而不是因为父进程仍持有读端而继续等待。Python 文档在替换 shell 管道的示例中明确强调了这一点。(docs.python.org)

11.1 不要只读取一个管道

下面的代码可能死锁：

process = Popen(
    ["noisy-command"],
    stdout=PIPE,
    stderr=PIPE,
    text=True,
)

stdout = process.stdout.read()
stderr = process.stderr.read()

如果子进程持续写 stderr，而父进程只读取 stdout，stderr 管道可能先被写满。子进程阻塞后不再写 stdout，父进程又在等待 stdout 结束，双方就无法继续。

正确方式：

process = Popen(
    ["noisy-command"],
    stdout=PIPE,
    stderr=PIPE,
    text=True,
)

stdout, stderr = process.communicate()

对于单个命令，通常直接使用：

result = subprocess.run(
    ["noisy-command"],
    capture_output=True,
    text=True,
)

更复杂的实时流处理才需要直接操作 Popen，而实时处理又会引入线程、异步 IO、背压和取消传播等问题。

12. Shell 管道和 Python 管道的故障差异

考虑这个命令：

producer | consumer

至少存在四种结果：

producer 成功，consumer 成功；
producer 失败，consumer 成功；
producer 成功，consumer 失败；
consumer 提前退出，producer 收到 SIGPIPE。

传统 shell 默认可能只报告最后一个命令的退出码。某些 shell 支持 pipefail，才会在管道中任意命令失败时报告失败。

Python 手动创建管道时，可以分别检查：

producer_rc = producer.wait()
consumer_rc = consumer.returncode

if consumer_rc != 0:
    return consumer_rc

if producer_rc != 0:
    return producer_rc

return 0

但“哪个错误优先”需要定义。例如，如果消费者因为找不到匹配内容而返回 1，这是否是错误，还是正常的“没有结果”？不同命令的语义不同：

grep 没有匹配项 → 常见约定为 1
grep 参数错误   → 2
grep 执行成功且有匹配 → 0

因此，管道的退出码策略不能仅由 Python 框架决定，而要由每个命令的契约决定。

13. `argparse`、管道和标准输入的组合

一个可管道化的命令通常支持：

tool process input.txt
tool process -
cat input.txt | tool process -

其中 - 是约定俗成的标准输入占位符。关键是不能无条件重复读取 stdin：

first = sys.stdin.read()
second = sys.stdin.read()

第二次读取通常只能得到空内容，因为标准输入是有当前位置的流，而不是可自动重置的字符串。

对于逐行处理，应使用流式方式：

for line in sys.stdin:
    process(line)

这样内存复杂度接近：

O(1)

更准确地说，它取决于单行最大长度和内部缓冲，而不是整个输入规模。

如果写成：

lines = sys.stdin.readlines()

内存复杂度接近：

O(n)

其中 $n$ 是输入总大小。对于小文件，这种写法简单；对于日志、数据库导出或无限流，它可能导致内存增长。

13.1 下游提前退出与 `BrokenPipeError`

执行：

python wr_cli.py filter INFO huge.log | head -n 1

head 读到一行后会关闭管道。上游继续写时，Python 可能收到 BrokenPipeError。

这通常不是业务失败，而是下游已经得到所需数据。可以在最外层处理：

try:
    return args.func(args)
except BrokenPipeError:
    return 0

不要在 BrokenPipeError 处理器中继续向同一个 stdout 打印错误，否则会再次触发异常。

某些 CLI 会进一步关闭标准输出：

try:
    ...
except BrokenPipeError:
    try:
        sys.stdout.close()
    finally:
        return 0

但关闭标准输出时还需注意解释器退出阶段的刷新行为。简单工具通常捕获并返回即可，复杂工具应通过端到端测试确认实际行为。

14. 输出格式：面向人和面向机器的接口不同

一个 CLI 的输出通常有两类：

人类可读输出
机器可解析输出

如果命令用于管道，输出格式应稳定：

print(json.dumps(result, ensure_ascii=False))

而不应把调试信息混入：

print("processing...")
print(json.dumps(result))

更合理的是：

print("processing...", file=sys.stderr)
print(json.dumps(result), file=sys.stdout)

对于结构化输出，可以把格式作为显式选项：

parser.add_argument(
    "--format",
    choices=("text", "json"),
    default="text",
)

业务层只产生数据对象：

result = {"count": 3}

输出层再决定格式：

def emit(result: dict[str, int], format_name: str) -> None:
    if format_name == "json":
        print(json.dumps(result, ensure_ascii=False))
    else:
        print(f"count: {result['count']}")

这使测试不必通过解析终端文本来验证核心业务。

15. 可测试性：把“进程边界”推到最外层

一个难以测试的 CLI 通常长这样：

def main():
    args = parser.parse_args()
    if args.command == "count":
        ...
        sys.exit(0)
    elif args.command == "filter":
        ...
        sys.exit(1)

问题在于：

解析器是全局对象，测试之间可能互相影响；
parse_args() 默认读取真实的 sys.argv；
业务逻辑和输出耦合；
sys.exit() 使函数不能自然返回；
文件和外部命令被直接执行；
测试必须依赖真实终端环境。

更可测试的生命周期是：

build_parser()
    ↓
parse_args(argv)
    ↓
Namespace → Config
    ↓
业务函数
    ↓
返回退出码
    ↓
最外层 SystemExit

其中只有最后一步触及进程退出。

15.1 直接测试解析器

import unittest


class ParserTests(unittest.TestCase):
    def test_count_arguments(self):
        parser = build_parser()

        args = parser.parse_args(
            ["count", "--unique", "input.txt"]
        )

        self.assertEqual(args.command, "count")
        self.assertTrue(args.unique)
        self.assertEqual(args.path, "input.txt")

    def test_run_arguments(self):
        parser = build_parser()

        args = parser.parse_args(
            ["run", "--timeout", "5", "echo", "hello"]
        )

        self.assertEqual(args.timeout, 5)
        self.assertEqual(args.program, "echo")
        self.assertEqual(args.args, ["hello"])

这里传入参数列表，而不是修改 sys.argv。这样每个测试拥有独立输入，且不会受到测试运行器参数的影响。

15.2 测试业务函数，而不是测试 `SystemExit`

from contextlib import redirect_stderr
from io import StringIO
from unittest import TestCase


class CountTests(TestCase):
    def test_count_stdin(self):
        old_stdin = sys.stdin
        try:
            sys.stdin = StringIO("a\n\nb\n")
            args = argparse.Namespace(
                path="-",
                unique=False,
            )
            self.assertEqual(run_count(args), 0)
        finally:
            sys.stdin = old_stdin

不过，直接替换全局 sys.stdin 会影响并发测试。更好的结构是让函数接收流：

def count_lines(
    stream: TextIO,
    *,
    unique: bool,
) -> int:
    if unique:
        return len({
            line.rstrip("\r\n")
            for line in stream
            if line.strip()
        })

    return sum(1 for line in stream if line.strip())

CLI 适配层只负责选择流：

def run_count(args: argparse.Namespace) -> int:
    stream, should_close = input_stream(args.path)
    try:
        print(count_lines(stream, unique=args.unique))
    finally:
        if should_close:
            stream.close()
    return 0

现在核心算法可以完全脱离进程、终端和文件系统测试：

from io import StringIO
import unittest


class CountLinesTests(unittest.TestCase):
    def test_count_non_empty_lines(self):
        stream = StringIO("a\n\nb\n")
        self.assertEqual(
            count_lines(stream, unique=False),
            2,
        )

    def test_count_unique_lines(self):
        stream = StringIO("a\na\nb\n")
        self.assertEqual(
            count_lines(stream, unique=True),
            2,
        )

15.3 测试错误与退出码

unittest.TestCase.assertRaises() 可以验证函数是否抛出指定异常，并支持上下文管理器形式。(docs.python.org)

class ParserErrorTests(unittest.TestCase):
    def test_invalid_workers(self):
        parser = build_parser()

        with self.assertRaises(SystemExit) as cm:
            parser.parse_args(
                ["run", "--timeout", "0", "echo"]
            )

        self.assertEqual(cm.exception.code, 2)

这里捕获的是 argparse 的命令行解析失败。由于默认 exit_on_error=True，无效参数时解析器会向 stderr 打印错误并以状态码 2 退出。(docs.python.org)

对于业务错误，推荐测试返回值：

class InputErrorTests(unittest.TestCase):
    def test_missing_file(self):
        args = argparse.Namespace(
            path="missing.txt",
            unique=False,
        )

        self.assertEqual(run_count(args), EX_NOINPUT)

如果希望完全不让 argparse 退出，可以使用：

parser = argparse.ArgumentParser(
    exit_on_error=False,
)

非法参数时捕获 argparse.ArgumentError。但这不能覆盖所有解析器行为；例如帮助动作本身仍然属于退出流程。官方文档说明，exit_on_error=False 用于手动捕获解析错误。(docs.python.org)

16. 测试真实 CLI：使用子进程验证最终契约

单元测试能验证函数，但无法完全证明：

包装脚本是否正确；
python -m package 是否可运行；
标准输入输出是否真正连接；
退出码是否传递到操作系统；
管道关闭时是否能正常结束；
编码和环境变量是否符合预期。

因此应补充少量端到端测试：

import subprocess
import sys
import unittest


class CliProcessTests(unittest.TestCase):
    def test_count_from_stdin(self):
        completed = subprocess.run(
            [
                sys.executable,
                "wr_cli.py",
                "count",
                "-",
            ],
            input="a\n\nb\n",
            text=True,
            capture_output=True,
            check=False,
        )

        self.assertEqual(completed.returncode, 0)
        self.assertEqual(completed.stdout, "2\n")
        self.assertEqual(completed.stderr, "")

subprocess.run() 的 input 会发送到子进程标准输入；当使用 text=True 时，输入可以是字符串。若 check=False，调用者直接检查 returncode；若 check=True，非零状态会转换为 CalledProcessError。(docs.python.org)

测试失败路径：

class CliFailureTests(unittest.TestCase):
    def test_missing_input_file(self):
        completed = subprocess.run(
            [
                sys.executable,
                "wr_cli.py",
                "count",
                "missing.txt",
            ],
            text=True,
            capture_output=True,
            check=False,
        )

        self.assertEqual(completed.returncode, EX_NOINPUT)
        self.assertIn("无法打开输入文件", completed.stderr)

这种测试成本比纯函数测试高，因为它启动了真实进程，但它验证的是使用者真正看到的接口。工程上通常不需要把所有测试都做成端到端测试，而是将测试分层：

大量纯函数测试
    ↓
少量 CLI 解析测试
    ↓
少量真实进程测试
    ↓
少量真实管道与外部命令测试

17. 依赖注入：让文件、进程和时钟可替换

可测试性不仅是“把 main() 拆开”，还要求将不稳定的外部依赖隔离。

17.1 把命令执行器作为依赖

from collections.abc import Callable
from typing import Any


RunCommand = Callable[..., subprocess.CompletedProcess[str]]


def run_external(
    command: list[str],
    *,
    runner: RunCommand = subprocess.run,
) -> int:
    try:
        result = runner(
            command,
            capture_output=True,
            text=True,
            check=False,
        )
    except FileNotFoundError:
        return EX_UNAVAILABLE

    return result.returncode

测试时使用假的执行器：

def fake_runner(*args: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]:
    return subprocess.CompletedProcess(
        args=args[0],
        returncode=7,
        stdout="",
        stderr="failed",
    )


class ExternalCommandTests(unittest.TestCase):
    def test_preserves_external_exit_code(self):
        rc = run_external(
            ["fake-command"],
            runner=fake_runner,
        )
        self.assertEqual(rc, 7)

这样测试不依赖机器上是否安装了某个程序，也不会真的修改系统。

17.2 把输出流作为参数

def emit_count(
    count: int,
    *,
    output: TextIO,
) -> None:
    print(count, file=output)

测试：

from io import StringIO


class OutputTests(unittest.TestCase):
    def test_emit_count(self):
        output = StringIO()

        emit_count(3, output=output)

        self.assertEqual(output.getvalue(), "3\n")

如果函数内部直接使用 print()，测试就需要捕获全局 sys.stdout；显式传入流能使数据流关系更清楚。

18. 解析器错误和业务错误必须有不同的处理位置

建议将生命周期拆成以下状态：

stateDiagram-v2
    [*] --> Parse
    Parse --> ParseError: 参数非法
    Parse --> Dispatch: 参数合法
    Dispatch --> Business
    Business --> Success: 返回 0
    Business --> DomainError: 输入或业务失败
    Business --> ExternalCommand
    ExternalCommand --> ExternalFailure: 非零退出码
    ExternalCommand --> Timeout: 超时
    ExternalCommand --> Success
    ParseError --> [*]: 退出 2
    DomainError --> [*]: 业务退出码
    ExternalFailure --> [*]: 映射或透传退出码
    Timeout --> [*]: 退出 124
    Success --> [*]: 退出 0

解析阶段失败：

wr count --unknown input.txt

调用者尚未进入业务逻辑，因此错误应由 argparse 负责。

业务阶段失败：

wr count missing.txt

语法是正确的，失败原因是资源不可用，因此由业务层返回输入错误退出码。

外部命令失败：

wr run false

命令成功启动，但 false 返回非零。此时不能把它误报为“找不到命令”；应根据工具契约选择透传 1、映射为统一错误码，或包装成更高层的失败。

这三个错误如果全部统一成：

except Exception:
    return 1

调用者就无法区分“命令写错”“输入不存在”“外部服务暂时不可用”和“程序本身有 bug”。

19. `parse_known_args()` 与未知参数透传

有些 CLI 需要把参数交给下游命令：

wr run python -m http.server 8000

如果使用普通的 parse_args()，解析器会尝试解释所有参数。对于明确的“本工具参数 + 下游剩余参数”模型，可以使用：

args, unknown = parser.parse_known_args(argv)

但这会降低错误检查强度：拼写错误的本工具参数也可能被当成下游参数。

例如，本来用户写错了：

wr run --timeuot 10 echo

如果程序无条件透传未知参数，错误可能直到 echo 或下游程序才暴露。

更清楚的设计是使用位置参数和 REMAINDER：

run_parser.add_argument("program")
run_parser.add_argument(
    "args",
    nargs=argparse.REMAINDER,
)

这表达了明确边界：

wr run 的选项
    ↓
program
    ↓
其余所有参数交给 program

但它也有一个语法限制：一旦进入 REMAINDER，后续内容通常不再由当前解析器解释。因此 wr run echo --timeout 3 中的 --timeout 会成为 echo 的参数，而不是 wr 的参数。

命令帮助应明确展示这个边界，避免让用户猜测选项属于哪一层。

20. 常见失败模式与诊断方法

20.1 在 `main()` 中直接读取全局 `sys.argv`

失败表现：

def main():
    args = parser.parse_args()

测试时无法自然传参，只能修改全局变量：

sys.argv = ["tool", "count", "-"]

改为：

def main(argv: Sequence[str] | None = None) -> int:
    args = parser.parse_args(argv)

生产运行时传入 None 仍会读取真实命令行，测试时可以直接传列表。

20.2 把业务输出和错误输出写到同一流

失败表现：

tool process input | jq .

jq 报 JSON 解析错误，因为上游打印了日志文本。

诊断方法：

tool process input >out.txt 2>err.txt

检查：

out.txt 是否只有机器数据
err.txt 是否只有诊断信息

20.3 使用 `stdout=PIPE` 却不消费输出

失败表现：程序在小输入上正常，大输入时卡住。

原因：子进程写满管道缓冲后阻塞，父进程没有同时读取。

修复：

subprocess.run(
    command,
    capture_output=True,
    text=True,
)

或：

process.communicate()

官方文档明确警告，使用 stdout=PIPE 或 stderr=PIPE 时不消费流可能造成死锁。(docs.python.org)

20.4 把 `FileNotFoundError` 和外部命令非零混为一谈

try:
    subprocess.run(["tool"], check=True)
except Exception:
    return 1

这段代码丢失了关键信息：

程序是否存在；
程序是否启动成功；
启动后返回了什么；
是否发生超时；
是否被信号终止。

应分别捕获：

except FileNotFoundError:
    ...
except subprocess.TimeoutExpired:
    ...
except subprocess.CalledProcessError as exc:
    ...

20.5 过度依赖 `shell=True`

失败表现：

用户输入导致命令注入；
空格路径解析错误；
Windows 与 POSIX 行为不同；
shell 返回码掩盖真实子命令状态。

诊断方法是先问：是否真的需要 shell 的语法能力？如果只需要执行一个程序，使用参数列表：

["program", arg1, arg2]

如果确实需要 shell，应限制输入来源，并为每种平台编写测试。

21. 与自动化脚本、重试、幂等和审计的衔接

CLI 经常成为自动化系统的入口，因此退出码必须与重试策略相容。

假设自动化平台规定：

退出码 0       → 成功，不重试
退出码 3       → 业务拒绝，不重试
退出码 5       → 临时依赖失败，可以重试
退出码 124     → 超时，可以重试
退出码 130     → 用户取消，不重试

那么 CLI 必须区分：

return 3  # 请求本身不符合业务规则
return 5  # 外部依赖暂时不可用
return 124  # 等待超时

如果所有失败都返回 1，调度器只能采用“所有错误都重试”或“所有错误都不重试”的粗粒度策略。

但返回“可重试”并不代表重试安全。若命令执行了部分写操作：

第一次调用：
    创建资源成功
    写入审计日志成功
    最后一步超时

第二次调用：
    再次创建资源

就可能产生重复资源。CLI 层至少应记录：

操作标识；
输入摘要；
目标资源；
开始和结束时间；
外部命令及其返回码；
是否发生超时；
是否已经产生副作用。

幂等性属于业务和存储设计，但退出码、审计日志和错误分类决定了自动化系统能否正确使用这种幂等性。

22. 与 Python 包构建和发布的衔接

当 CLI 作为 Python 包发布时，用户可能通过三种方式启动它：

python script.py
python -m package
wr

这三种入口的 sys.argv[0]、模块上下文和帮助中的程序名可能不同。Python 3.14 的 ArgumentParser 默认 prog 会根据程序实际执行方式反映入口，而不再总是简单取 os.path.basename(sys.argv[0])。(docs.python.org)

如果帮助信息必须固定，可以显式指定：

parser = argparse.ArgumentParser(prog="wr")

打包时应测试最终产物，而不只是源码目录中的运行方式：

源码运行：
    python wr_cli.py --help

模块运行：
    python -m wr --help

安装后的脚本：
    wr --help

尤其要验证：

wheel 安装后入口是否存在；
sdist 安装后数据文件是否完整；
当前工作目录变化后，程序是否仍能找到资源；
退出码是否在包装脚本中保持不变；
帮助中的 prog 是否符合发布名称。

CLI 入口不应依赖：

Path(__file__).parent / "some-relative-file"

来推断用户当前目录下的输入。输入路径、配置路径和程序资源路径应明确区分。

23. 一个稳定 CLI 的最小结构

对于中小型项目，下面的结构已经足够表达清晰边界：

package/
    __init__.py
    cli.py          # build_parser、main、异常到退出码的映射
    commands.py     # 子命令处理器
    core.py         # 不依赖终端的核心逻辑
    process.py      # subprocess 封装
tests/
    test_parser.py
    test_core.py
    test_cli_process.py

职责划分：

cli.py
    argv → Namespace → handler → exit code

commands.py
    参数对象 → 输入输出适配 → 业务调用

core.py
    纯数据处理、规则判断、状态转换

process.py
    外部进程、超时、信号和返回码处理

这个结构不是必须的目录规范，但它体现了一个关键原则：

CLI 是适配层，不应成为所有逻辑的容器。

当参数解析、文件操作、网络请求、重试、子进程和输出格式全部堆在一个 main() 中，代码即使暂时能运行，也会难以区分错误来源，更难验证管道和退出码契约。

24. 结语：把 CLI 当作协议实现

一个可靠的 Python CLI 不是“能从命令行启动的脚本”，而是一个具有明确协议的进程：

输入协议：
    argv、stdin、环境变量、文件

处理协议：
    解析、分派、业务执行、外部进程管理

输出协议：
    stdout 数据
    stderr 诊断
    exit code 状态

argparse 负责把参数语法转换为结构化对象；子命令负责划分独立的命令空间；subprocess 负责连接外部进程；管道负责传递流式数据；退出码负责向自动化调用者报告状态；可测试性则要求把这些边界显式保留，而不是隐藏在全局状态和深层 sys.exit() 中。

当这些职责被分开后，CLI 才能同时满足三个条件：

人可以直接使用
程序可以通过管道组合
测试可以不依赖真实副作用

这也是从一次性自动化脚本走向可维护命令行工具的分界线。

系列导航与关联阅读

官方资料

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。

Python CLI 工程：argparse、子命令、退出码、管道和可测试性

1. CLI 的真实边界：参数、数据流和进程状态

2. argparse 的职责：把文本参数转换为结构化配置

2.1 parse_args() 不是业务逻辑

2.2 type 是转换，不是完整验证

2.3 默认值必须区分“未提供”和“显式提供”

3. 参数解析的三个层次：语法、类型和语义

第一层：语法合法

第二层：类型合法

第三层：业务语义合法

4. 子命令：把一个程序拆成多个独立语法空间

5. 一个完整的多子命令工具

6. required=True 与没有子命令的状态

7. Python 3.14 的 argparse 能力与兼容性边界

8. 退出码：进程如何向调用者报告结果

8.1 return 和 sys.exit() 的区别

8.2 退出码的设计原则

8.3 sys.exit("message") 的陷阱

9. 外部命令：subprocess.run()、返回值和异常

9.1 不要把 shell 命令字符串当作参数序列

10. 超时、信号和负退出码

11. 管道的本质：前一个进程的 stdout 连接后一个进程的 stdin

11.1 不要只读取一个管道

12. Shell 管道和 Python 管道的故障差异

13. argparse、管道和标准输入的组合

13.1 下游提前退出与 BrokenPipeError

14. 输出格式：面向人和面向机器的接口不同

15. 可测试性：把“进程边界”推到最外层

15.1 直接测试解析器

15.2 测试业务函数，而不是测试 SystemExit

15.3 测试错误与退出码

16. 测试真实 CLI：使用子进程验证最终契约

17. 依赖注入：让文件、进程和时钟可替换

17.1 把命令执行器作为依赖

17.2 把输出流作为参数

18. 解析器错误和业务错误必须有不同的处理位置

19. parse_known_args() 与未知参数透传

20. 常见失败模式与诊断方法

20.1 在 main() 中直接读取全局 sys.argv

20.2 把业务输出和错误输出写到同一流

20.3 使用 stdout=PIPE 却不消费输出

20.4 把 FileNotFoundError 和外部命令非零混为一谈

20.5 过度依赖 shell=True

21. 与自动化脚本、重试、幂等和审计的衔接

22. 与 Python 包构建和发布的衔接

23. 一个稳定 CLI 的最小结构

24. 结语：把 CLI 当作协议实现

系列导航与关联阅读

官方资料

2. `argparse` 的职责：把文本参数转换为结构化配置

2.1 `parse_args()` 不是业务逻辑

2.2 `type` 是转换，不是完整验证

6. `required=True` 与没有子命令的状态

7. Python 3.14 的 `argparse` 能力与兼容性边界

8.1 `return` 和 `sys.exit()` 的区别

8.3 `sys.exit("message")` 的陷阱

9. 外部命令：`subprocess.run()`、返回值和异常

11. 管道的本质：前一个进程的 `stdout` 连接后一个进程的 `stdin`

13. `argparse`、管道和标准输入的组合

13.1 下游提前退出与 `BrokenPipeError`

15.2 测试业务函数，而不是测试 `SystemExit`

19. `parse_known_args()` 与未知参数透传

20.1 在 `main()` 中直接读取全局 `sys.argv`

20.3 使用 `stdout=PIPE` 却不消费输出

20.4 把 `FileNotFoundError` 和外部命令非零混为一谈

20.5 过度依赖 `shell=True`