Python CLI 应用
大约 12 分钟约 3656 字
Python CLI 应用
简介
Python CLI（命令行接口）应用是通过终端交互的工具程序，广泛用于自动化脚本、数据处理、运维工具和开发工具链。Python 标准库提供 argparse，第三方库 click 和 typer 则提供更现代的 CLI 构建体验。
特点
实现
argparse 标准库 CLI
"""文件处理 CLI 工具"""
import argparse
import sys
import json
from pathlib import Path
def process_file(input_path: Path, output_path: Path, format: str, verbose: bool):
"""处理文件的核心逻辑"""
if verbose:
print(f"读取文件: {input_path}")
data = json.loads(input_path.read_text(encoding="utf-8"))
if format == "csv":
import csv
with open(output_path, "w", newline="", encoding="utf-8") as f:
if isinstance(data, list) and data:
writer = csv.DictWriter(f, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
elif format == "yaml":
import yaml
output_path.write_text(yaml.dump(data, allow_unicode=True), encoding="utf-8")
else:
output_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
if verbose:
print(f"输出文件: {output_path}")
def main():
parser = argparse.ArgumentParser(
prog="filetool",
description="JSON 文件格式转换工具",
epilog="示例: python cli.py convert input.json -o output.csv -f csv --verbose",
)
subparsers = parser.add_subparsers(dest="command", help="可用命令")
# convert 子命令
convert_parser = subparsers.add_parser("convert", help="转换文件格式")
convert_parser.add_argument("input", type=Path, help="输入文件路径")
convert_parser.add_argument("-o", "--output", type=Path, help="输出文件路径")
convert_parser.add_argument(
"-f", "--format",
choices=["json", "csv", "yaml"],
default="json",
help="输出格式 (默认: json)",
)
convert_parser.add_argument("-v", "--verbose", action="store_true", help="详细输出")
# validate 子命令
validate_parser = subparsers.add_parser("validate", help="校验 JSON 文件")
validate_parser.add_argument("file", type=Path, help="待校验的 JSON 文件")
validate_parser.add_argument("--schema", type=Path, help="JSON Schema 文件路径")
args = parser.parse_args()
if args.command == "convert":
output = args.output or args.input.with_suffix(f".{args.format}")
process_file(args.input, output, args.format, args.verbose)
elif args.command == "validate":
try:
json.loads(args.file.read_text(encoding="utf-8"))
print(f"校验通过: {args.file}")
except json.JSONDecodeError as e:
print(f"校验失败: {e}", file=sys.stderr)
sys.exit(1)
else:
parser.print_help()
if __name__ == "__main__":
main()click 现代化 CLI 框架
"""用户管理 CLI 工具"""
import click
import json
from pathlib import Path
from datetime import datetime
USERS_FILE = Path("users.json")
def load_users() -> list[dict]:
if USERS_FILE.exists():
return json.loads(USERS_FILE.read_text(encoding="utf-8"))
return []
def save_users(users: list[dict]) -> None:
USERS_FILE.write_text(json.dumps(users, indent=2, ensure_ascii=False), encoding="utf-8")
@click.group()
@click.version_option(version="1.0.0")
def cli():
"""用户管理命令行工具"""
pass
@cli.command()
@click.argument("name")
@click.option("--email", "-e", required=True, help="用户邮箱")
@click.option("--role", "-r", default="user", type=click.Choice(["admin", "user", "guest"]))
@click.option("--age", "-a", type=int, help="用户年龄")
def add(name: str, email: str, role: str, age: int | None):
"""添加新用户"""
users = load_users()
user_id = len(users) + 1
user = {
"id": user_id,
"name": name,
"email": email,
"role": role,
"age": age,
"created_at": datetime.now().isoformat(),
}
users.append(user)
save_users(users)
click.echo(click.style(f"用户添加成功: ID={user_id}, name={name}", fg="green"))
@cli.command()
@click.option("--role", "-r", help="按角色筛选")
@click.option("--format", "fmt", type=click.Choice(["table", "json"]), default="table")
def list_users(role: str | None, fmt: str):
"""列出所有用户"""
users = load_users()
if role:
users = [u for u in users if u.get("role") == role]
if not users:
click.echo("没有找到用户")
return
if fmt == "json":
click.echo(json.dumps(users, indent=2, ensure_ascii=False))
else:
for u in users:
click.echo(f" ID={u['id']} | {u['name']} | {u['email']} | {u['role']}")
@cli.command()
@click.argument("user_id", type=int)
@click.confirmation_option(prompt="确定删除该用户？")
def remove(user_id: int):
"""删除用户"""
users = load_users()
new_users = [u for u in users if u["id"] != user_id]
if len(new_users) == len(users):
click.echo(click.style(f"用户 ID={user_id} 不存在", fg="red"))
return
save_users(new_users)
click.echo(click.style(f"用户 ID={user_id} 已删除", fg="green"))
@cli.command()
@click.argument("user_id", type=int)
@click.option("--name", "-n", help="修改用户名")
@click.option("--email", "-e", help="修改邮箱")
def update(user_id: int, name: str | None, email: str | None):
"""更新用户信息"""
users = load_users()
for u in users:
if u["id"] == user_id:
if name:
u["name"] = name
if email:
u["email"] = email
save_users(users)
click.echo(click.style(f"用户 ID={user_id} 已更新", fg="green"))
return
click.echo(click.style(f"用户 ID={user_id} 不存在", fg="red"))
if __name__ == "__main__":
cli()typer 类型驱动的 CLI
"""数据处理 CLI 工具"""
import typer
from typing import Optional
from pathlib import Path
from enum import Enum
app = typer.Typer(help="数据处理命令行工具")
class OutputFormat(str, Enum):
csv = "csv"
json = "json"
excel = "excel"
@app.command()
def process(
input_file: Path = typer.Argument(..., help="输入文件路径", exists=True),
output: Optional[Path] = typer.Option(None, "--output", "-o", help="输出文件路径"),
format: OutputFormat = typer.Option(OutputFormat.csv, "--format", "-f", help="输出格式"),
batch_size: int = typer.Option(1000, "--batch", "-b", help="批处理大小", min=1, max=10000),
dry_run: bool = typer.Option(False, "--dry-run", help="试运行模式，不写入文件"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
):
"""处理数据文件"""
if verbose:
typer.echo(f"输入文件: {input_file}")
typer.echo(f"输出格式: {format.value}")
typer.echo(f"批处理大小: {batch_size}")
total_lines = sum(1 for _ in open(input_file, encoding="utf-8"))
typer.echo(f"文件共 {total_lines} 行")
if dry_run:
typer.echo(typer.style("试运行模式，跳过实际处理", fg=typer.colors.YELLOW))
raise typer.Exit()
output_path = output or input_file.with_suffix(f".{format.value}")
typer.echo(typer.style(f"处理完成: {output_path}", fg=typer.colors.GREEN))
@app.command()
def validate(
files: list[Path] = typer.Argument(..., help="待校验的文件列表"),
strict: bool = typer.Option(False, "--strict", help="严格模式"),
):
"""批量校验文件"""
results = {"valid": 0, "invalid": 0}
for f in files:
if f.exists() and f.suffix == ".json":
import json
try:
json.loads(f.read_text(encoding="utf-8"))
typer.echo(typer.style(f" [OK] {f}", fg=typer.colors.GREEN))
results["valid"] += 1
except json.JSONDecodeError as e:
typer.echo(typer.style(f" [FAIL] {f}: {e}", fg=typer.colors.RED))
results["invalid"] += 1
if strict:
raise typer.Exit(code=1)
else:
typer.echo(typer.style(f" [SKIP] {f}", fg=typer.colors.YELLOW))
typer.echo(f"\n校验完成: {results['valid']} 通过, {results['invalid']} 失败")
@app.command()
def info():
"""显示系统信息"""
import sys
import platform
typer.echo(f"Python: {sys.version}")
typer.echo(f"平台: {platform.platform()}")
typer.echo(f"编码: {sys.getdefaultencoding()}")
if __name__ == "__main__":
app()进度条与终端美化（rich）
"""带进度条和美化的 CLI 工具"""
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, BarColumn, TextColumn
from rich.table import Table
from rich.panel import Panel
from rich import print as rprint
import time
import random
console = Console()
def show_progress():
"""带进度条的批量处理"""
tasks = [f"task_{i}" for i in range(50)]
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
console=console,
) as progress:
task = progress.add_task("处理数据...", total=len(tasks))
for task_name in tasks:
time.sleep(0.02) # 模拟处理
progress.advance(task)
console.print("[bold green]处理完成！[/bold green]")
def show_table():
"""表格展示数据"""
table = Table(title="用户列表", show_header=True, header_style="bold magenta")
table.add_column("ID", style="dim", width=6)
table.add_column("姓名", min_width=10)
table.add_column("邮箱", min_width=20)
table.add_column("状态", min_width=8)
users = [
{"id": 1, "name": "张三", "email": "zhang@example.com", "status": "活跃"},
{"id": 2, "name": "李四", "email": "li@example.com", "status": "停用"},
{"id": 3, "name": "王五", "email": "wang@example.com", "status": "活跃"},
]
for u in users:
status_style = "green" if u["status"] == "活跃" else "red"
table.add_row(
str(u["id"]),
u["name"],
u["email"],
f"[{status_style}]{u['status']}[/{status_style}]",
)
console.print(table)
def show_summary():
"""面板展示摘要"""
console.print(Panel.fit(
"[bold blue]处理报告[/bold blue]\n\n"
"总记录数: [green]1,234[/green]\n"
"成功: [green]1,200[/green]\n"
"失败: [red]34[/red]\n"
"耗时: [yellow]12.5s[/yellow]",
title="数据导入",
border_style="blue",
))
# show_progress()
# show_table()
# show_summary()CLI 配置管理与日志
生产级 CLI 工具需要完善的配置管理和日志体系。以下展示如何实现多层级配置和结构化日志。
"""CLI 配置管理"""
import os
import json
import logging
from pathlib import Path
from dataclasses import dataclass, field, asdict
from typing import Optional
@dataclass
class CLIConfig:
"""CLI 配置数据类"""
api_url: str = "https://api.example.com"
timeout: int = 30
max_retries: int = 3
output_dir: Path = field(default_factory=lambda: Path("./output"))
log_level: str = "INFO"
dry_run: bool = False
@classmethod
def load(cls, config_path: Optional[Path] = None) -> "CLIConfig":
"""多层级加载配置：默认值 < 配置文件 < 环境变量 < 命令行参数"""
config = cls()
# 层级 1：项目配置文件
project_config = Path("cli_config.json")
if project_config.exists():
config._merge_file(project_config)
# 层级 2：用户目录配置
user_config = Path.home() / ".mycli" / "config.json"
if user_config.exists():
config._merge_file(user_config)
# 层级 3：指定的配置文件
if config_path and config_path.exists():
config._merge_file(config_path)
# 层级 4：环境变量覆盖
if url := os.getenv("MYCLI_API_URL"):
config.api_url = url
if timeout := os.getenv("MYCLI_TIMEOUT"):
config.timeout = int(timeout)
return config
def _merge_file(self, path: Path):
"""合并 JSON 配置文件"""
data = json.loads(path.read_text(encoding="utf-8"))
for key, value in data.items():
if hasattr(self, key):
if key == "output_dir":
setattr(self, key, Path(value))
else:
setattr(self, key, value)
def save(self, path: Path):
"""保存配置到文件"""
path.parent.mkdir(parents=True, exist_ok=True)
data = {k: str(v) if isinstance(v, Path) else v for k, v in asdict(self).items()}
path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
def setup_logging(level: str = "INFO", log_file: Optional[Path] = None):
"""配置 CLI 日志"""
logger = logging.getLogger("mycli")
logger.setLevel(getattr(logging, level.upper()))
# 控制台输出（简洁格式）
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter(
"%(levelname)s: %(message)s"
))
logger.addHandler(console_handler)
# 文件输出（详细格式）
if log_file:
log_file.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(log_file, encoding="utf-8")
file_handler.setFormatter(logging.Formatter(
"%(asctime)s [%(levelname)s] %(name)s - %(message)s"
))
logger.addHandler(file_handler)
return loggerCLI 打包与分发
# pyproject.toml 配置示例
# [project]
# name = "mycli"
# version = "1.0.0"
# dependencies = ["click>=8.0", "rich>=13.0", "requests>=2.28"]
#
# [project.scripts]
# mycli = "mycli.main:cli" # 注册 CLI 入口点
#
# [project.optional-dependencies]
# dev = ["pytest", "pytest-cov"]
# 使用方式：
# pip install -e . # 开发模式安装
# mycli --help # 直接使用命令
# pip install build && python -m build # 构建分发包
# PyInstaller 打包为独立可执行文件
# pip install pyinstaller
# pyinstaller --onefile --name mycli mycli/main.py
# pyinstaller --onefile --windowed --name mycli-gui mycli/gui.py # GUI 模式
# 常用 PyInstaller 参数：
# --onefile 打包为单个可执行文件
# --clean 清理临时文件
# --add-data 添加非 Python 文件
# --hidden-import 显式指定隐式导入
# --icon app.ico 设置图标实战：完整的 CLI 项目结构
mycli/
├── pyproject.toml # 项目配置和 CLI 入口点
├── README.md
├── src/
│ └── mycli/
│ ├── __init__.py
│ ├── main.py # CLI 入口，注册子命令
│ ├── config.py # 配置管理
│ ├── commands/ # 子命令模块
│ │ ├── __init__.py
│ │ ├── data.py # data 子命令（导入/导出/校验）
│ │ ├── user.py # user 子命令（增删改查）
│ │ └── system.py # system 子命令（信息/状态）
│ ├── services/ # 业务逻辑层
│ │ ├── api.py # API 客户端
│ │ └── database.py # 数据库操作
│ └── utils/ # 工具函数
│ ├── output.py # rich 输出封装
│ └── validation.py# 校验工具
├── tests/
│ ├── test_commands.py
│ └── test_services.py
└── scripts/
└── build.sh # 构建脚本# main.py — CLI 入口点
import click
from mycli.commands.data import data_cmd
from mycli.commands.user import user_cmd
from mycli.commands.system import system_cmd
@click.group()
@click.version_option(version="1.0.0", prog_name="mycli")
@click.option("--config", "-c", type=click.Path(), help="配置文件路径")
@click.option("--verbose", "-v", is_flag=True, help="详细输出")
@click.pass_context
def cli(ctx, config, verbose):
"""多功能命令行工具"""
ctx.ensure_object(dict)
ctx.obj["config_path"] = config
ctx.obj["verbose"] = verbose
cli.add_command(data_cmd, "data")
cli.add_command(user_cmd, "user")
cli.add_command(system_cmd, "system")
# 测试 CLI 命令
# mycli data import --file data.json --format csv
# mycli user list --role admin --format table
# mycli system info优点
缺点
总结
Python CLI 应用是连接开发效率与自动化运维的桥梁。简单工具用 argparse 即可，中等复杂度推荐 click，追求现代体验和类型安全选 typer。配合 rich 库可实现专业级的终端输出效果。在项目中将常用操作封装为 CLI 命令，比维护散落的脚本更利于团队协作。
关键知识点
- argparse 是标准库方案，无需安装但 API 较繁琐
- click 通过装饰器定义命令和参数，代码更简洁
- typer 基于类型注解自动生成 CLI，与 FastAPI 风格一致
- rich 提供进度条、表格、高亮等终端美化能力
CLI 测试策略
CLI 工具的测试需要覆盖参数解析、命令执行和输出格式等多个维度。
"""CLI 测试示例"""
import pytest
from click.testing import CliRunner
from mycli.main import cli
@pytest.fixture
def runner():
"""Click 测试运行器"""
return CliRunner()
class TestUserCommands:
def test_add_user(self, runner):
"""测试添加用户"""
result = runner.invoke(cli, [
"user", "add", "张三",
"--email", "zhang@example.com",
"--role", "admin"
])
assert result.exit_code == 0
assert "张三" in result.output
def test_add_user_missing_email(self, runner):
"""测试缺少必填参数"""
result = runner.invoke(cli, ["user", "add", "张三"])
assert result.exit_code != 0
assert "Missing option" in result.output or "email" in result.output
def test_list_users_empty(self, runner):
"""测试空用户列表"""
result = runner.invoke(cli, ["user", "list"])
assert result.exit_code == 0
assert "没有找到用户" in result.output
def test_list_users_json_format(self, runner):
"""测试 JSON 格式输出"""
# 先添加用户
runner.invoke(cli, [
"user", "add", "李四",
"--email", "li@example.com"
])
result = runner.invoke(cli, ["user", "list", "--format", "json"])
assert result.exit_code == 0
import json
data = json.loads(result.output)
assert isinstance(data, list)
def test_remove_user_with_confirmation(self, runner):
"""测试带确认的删除操作"""
result = runner.invoke(cli, ["user", "remove", "1"], input="y\n")
assert result.exit_code == 0
def test_invalid_role(self, runner):
"""测试无效的角色选项"""
result = runner.invoke(cli, [
"user", "add", "测试",
"--email", "test@example.com",
"--role", "invalid"
])
assert result.exit_code != 0
class TestDataCommands:
def test_validate_valid_json(self, runner, tmp_path):
"""测试校验有效 JSON 文件"""
json_file = tmp_path / "test.json"
json_file.write_text('{"key": "value"}', encoding="utf-8")
result = runner.invoke(cli, ["data", "validate", str(json_file)])
assert "OK" in result.output
def test_validate_invalid_json(self, runner, tmp_path):
"""测试校验无效 JSON 文件"""
json_file = tmp_path / "bad.json"
json_file.write_text("{invalid}", encoding="utf-8")
result = runner.invoke(cli, ["data", "validate", str(json_file)])
assert "FAIL" in result.output
def test_dry_run_mode(self, runner, tmp_path):
"""测试试运行模式"""
input_file = tmp_path / "data.json"
input_file.write_text('[]', encoding="utf-8")
result = runner.invoke(cli, [
"data", "process", str(input_file), "--dry-run"
])
assert "试运行" in result.output
# 试运行不应生成输出文件
output_file = tmp_path / "data.csv"
assert not output_file.exists()CLI 性能与管道集成
"""管道友好的 CLI 工具设计"""
import sys
import json
import select
from pathlib import Path
def read_from_stdin_or_file(filepath: str | None = None):
"""支持从标准输入或文件读取数据"""
if filepath:
return Path(filepath).read_text(encoding="utf-8")
# 检查是否有管道输入
if not sys.stdin.isatty():
return sys.stdin.read()
return None
def streaming_process(input_path: Path, batch_size: int = 1000):
"""流式处理大文件，避免一次性加载到内存"""
import csv
with open(input_path, encoding="utf-8") as f:
reader = csv.DictReader(f)
batch = []
for row in reader:
batch.append(row)
if len(batch) >= batch_size:
yield batch
batch = []
if batch:
yield batch
# 使用示例：
# cat data.json | python mycli.py process -
# python mycli.py process --file data.json
# python mycli.py process < data.json
# python mycli.py process --file bigdata.csv --batch-size 5000项目落地视角
- 将项目中的数据迁移、批量处理、运维操作封装为 CLI 子命令
- CLI 工具统一使用 --verbose/--debug 控制日志输出级别
- 使用 pyproject.toml 的 [project.scripts] 注册 CLI 入口点
- CI/CD 流水线中通过 CLI 工具执行部署、校验和回滚操作
常见误区
- 使用 sys.argv 手动解析参数，不利用成熟的 CLI 框架
- CLI 工具缺少 --help 和版本号信息
- 在 CLI 中硬编码配置路径，不支持通过参数或环境变量覆盖
- 缺少错误处理，异常时输出 Python 堆栈信息而非友好提示
进阶路线
- 学习 typer + rich 的组合使用，构建专业级 CLI 体验
- 研究 prompt-toolkit 实现交互式 CLI（如 IPython、mycli）
- 了解 Python CLI 应用的打包和分发方式（PyInstaller、nuitka）
- 探索 Textual 框架构建终端 TUI 应用
适用场景
- 数据处理、文件转换、批量导入导出等运维工具
- CI/CD 流水线中的构建、部署、校验脚本
- 开发团队的内部工具和代码生成器
落地建议
- 统一使用 typer 或 click 作为 CLI 框架，保持团队风格一致
- 每个 CLI 命令提供 --help 和使用示例
- 错误时输出友好的提示信息而非 Python 异常堆栈
排错清单
- 检查参数类型转换是否正确（路径、数字、枚举）
- 确认子命令的注册和调用是否正确
- 排查编码问题导致的中文输出乱码
复盘问题
- 你的项目中有哪些散落的脚本可以统一为 CLI 命令？
- CLI 工具的帮助信息是否足够清晰？新用户能否快速上手？
- 错误处理是否友好？用户能否根据提示自行解决问题？