饮墨

子安饮墨馀三斗,留与卿儿作赋来

用 Python Click 构建专业运维 CLI 工具:从散装脚本到统一命令行,附 3 个实战场景

痛点:运维脚本越来越多,管理越来越乱

团队服务器从 20 台涨到 200 台,运维脚本也从 5 个涨到了 50 个——check_disk.shrestart_service.pysync_config.sh……散落在各个目录,参数靠口口相传,新人来了不知道该跑哪个、怎么传参。

更现实的问题:

  • argparse 写复杂子命令时代码又臭又长
  • 脚本没有统一入口,每次都要 find + grep 找脚本
  • 参数校验靠 if-else,传错了直接报 traceback

你需要的不是更多脚本,而是一个统一的运维 CLI 工具——像 kubectldocker 那样,子命令清晰、参数自动补全、--help 就能看懂。

方案:用 Click 框架 30 分钟搭一套运维 CLI

Click 是 Python 最成熟的 CLI 框架(Flask 作者开发),核心优势:

对比项 argparse Click
子命令支持 手动拼 subparsers 装饰器一行搞定
参数校验 自己写 if-else 内置 type/callback
帮助文档 手写 help 字符串 自动生成
嵌套命令组 痛苦 @click.group() 原生支持
输出美化 自己搞 click.echo / click.style

一句话:argparse 是手动挡,Click 是自动挡

实操:3 个场景从零构建 ops-cli

第 1 步:安装 + 项目骨架

pip install click rich
mkdir -p ops-cli && cd ops-cli
touch ops.py

基础骨架 ops.py

import click

@click.group()
@click.version_option(version="1.0.0")
def cli():
    """OPS-CLI:团队统一运维命令行工具"""
    pass

if __name__ == "__main__":
    cli()

运行 python ops.py --help 就能看到自动生成的帮助文档。

第 2 步:场景一——批量检查磁盘空间

import subprocess
import click

@cli.command()
@click.option("--threshold", "-t", default=80, help="告警阈值(%)")
@click.argument("hosts", nargs=-1, required=True)
def disk_check(threshold, hosts):
    """批量检查远程主机磁盘使用率"""
    for host in hosts:
        click.echo(f"\n{'='*40}")
        click.echo(click.style(f"📍 {host}", fg="cyan", bold=True))
        try:
            result = subprocess.run(
                ["ssh", "-o", "ConnectTimeout=5", host,
                 "df -h --output=target,pcent / | tail -1"],
                capture_output=True, text=True, timeout=10
            )
            if result.returncode != 0:
                click.echo(click.style(f"  ❌ 连接失败: {result.stderr.strip()}", fg="red"))
                continue
            usage = int(result.stdout.strip().split()[-1].replace("%", ""))
            color = "red" if usage >= threshold else "green"
            click.echo(click.style(f"  磁盘使用率: {usage}%", fg=color))
            if usage >= threshold:
                click.echo(click.style(f"  ⚠️  超过阈值 {threshold}%!", fg="yellow"))
        except subprocess.TimeoutExpired:
            click.echo(click.style("  ❌ 超时", fg="red"))

使用方式:

python ops.py disk-check -t 85 web-01 web-02 db-01

第 3 步:场景二——服务批量重启(带确认)

@cli.command()
@click.option("--service", "-s", required=True, help="systemd 服务名")
@click.option("--hosts-file", "-f", type=click.Path(exists=True), help="主机列表文件")
@click.option("--yes", "-y", is_flag=True, help="跳过确认")
def restart(service, hosts_file, yes):
    """批量重启指定 systemd 服务"""
    with open(hosts_file) as f:
        hosts = [line.strip() for line in f if line.strip() and not line.startswith("#")]

    click.echo(f"即将在 {len(hosts)} 台主机上重启 {service}")
    if not yes:
        click.confirm("确认继续?", abort=True)  # 自动处理 y/n,n 直接退出

    for host in hosts:
        click.echo(f"  🔄 {host}: ", nl=False)
        result = subprocess.run(
            ["ssh", host, f"sudo systemctl restart {service} && "
             f"systemctl is-active {service}"],
            capture_output=True, text=True, timeout=30
        )
        if "active" in result.stdout:
            click.echo(click.style("✅ running", fg="green"))
        else:
            click.echo(click.style(f"❌ {result.stderr.strip()}", fg="red"))

使用方式:

# 交互确认
python ops.py restart -s nginx -f hosts.txt

# CI/CD 里跳过确认
python ops.py restart -s nginx -f hosts.txt -y

第 4 步:场景三——快速查看服务状态 Dashboard

from datetime import datetime

@cli.command()
@click.option("--services", "-s", multiple=True,
              default=["nginx", "docker", "prometheus-node-exporter"],
              help="要检查的服务列表")
@click.argument("host")
def status(services, host):
    """查看单台主机核心服务状态"""
    click.echo(f"\n📊 {host} 服务状态 — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
    click.echo("-" * 45)
    for svc in services:
        result = subprocess.run(
            ["ssh", "-o", "ConnectTimeout=5", host,
             f"systemctl is-active {svc} 2>/dev/null || echo inactive"],
            capture_output=True, text=True, timeout=10
        )
        state = result.stdout.strip()
        icon = "🟢" if state == "active" else "🔴"
        click.echo(f"  {icon} {svc:<35} {state}")

使用方式:

python ops.py status web-01
python ops.py status -s nginx -s redis -s mysql db-01

第 5 步:打包成系统命令

setup.pypyproject.toml 中配置 entry_points:

# pyproject.toml
[project.scripts]
ops = "ops:cli"
pip install -e .
ops disk-check -t 90 web-01 web-02  # 像系统命令一样直接用
ops --help                            # 自动列出所有子命令

避坑:3 个常见问题

坑 1:子命令函数名里的下划线变成了连字符

Click 会自动把函数名 disk_check 转为命令名 disk-check。如果你想保持下划线:

@click.group(context_settings={"token_normalize_func": lambda x: x})

但建议顺其自然——连字符才是 CLI 惯例(kubectl get-pods 不对,但 ops disk-check 没问题)。

坑 2:在 Crontab / CI 环境里报 encoding 错误

Click 依赖终端的 locale 设置。在非交互环境中加:

export LC_ALL=C.UTF-8
export LANG=C.UTF-8

或者在代码里设置:

import os
os.environ.setdefault("LC_ALL", "C.UTF-8")

坑 3:参数太多导致命令行超长

当一个命令需要十几个参数时,用 YAML/JSON 配置文件代替:

@cli.command()
@click.option("--config", "-c", type=click.Path(exists=True), required=True)
def deploy(config):
    """按配置文件执行部署"""
    import yaml
    with open(config) as f:
        conf = yaml.safe_load(f)
    # 从 conf 里读取所有参数

总结

做法 结果
50 个散装脚本 新人懵,老人也忘参数
统一 ops-cli 一个入口、自动补全、--help 即文档

核心结论:当运维脚本超过 10 个,就该用 Click 把它们收编成一个 CLI 工具。投入半天,换来的是整个团队的效率提升。

进阶方向:加上 click-completion 实现 Tab 自动补全,配合 Rich 做表格输出,再用 click.testing.CliRunner 写单元测试——运维工具也能有工程质量。

您还没有登录,请登录后发表评论。