Python 运维自动化¶

Python 自动化适合处理 Ansible 不方便表达的流程，例如读取多个接口、转换数据、生成报表、批量校验配置、调用不同系统并汇总结果。涉及大规模系统配置时优先使用 Ansible；Python 更适合编排、数据转换和定制逻辑。

读取配置与目标清单
  → 参数校验
  → 逐个/并发执行 API、SSH 或文件操作
  → 超时、重试与错误分类
  → 输出结果和失败明细
  → 返回明确退出码

项目准备¶

uv init --no-package ops-automation
cd ops-automation
uv python pin 3.12
uv add requests pyyaml paramiko tenacity
uv add --dev pytest ruff
uv run python main.py --help

推荐目录：

ops-automation/
├── pyproject.toml
├── uv.lock
├── main.py
├── config.example.yml
├── src/
│   ├── api.py
│   ├── ssh.py
│   └── report.py
├── tests/
├── input/
└── output/

真实密码、私钥、Token 和生产主机清单不要放进公共仓库；提交脱敏的示例配置，运行时通过环境变量、只读挂载或密钥系统注入。

脚本必须具备的运维能力¶

能力	原因
命令行参数	明确输入、环境、并发数、输出目录和 dry-run
超时	防止一个接口或主机永久卡住整个批次
有限重试	处理临时网络错误，但不放大认证失败和业务错误
幂等	重复执行不会重复创建、重复写入或破坏已有状态
日志	记录批次、目标、动作、耗时与错误，不泄露秘密
失败明细	成功和失败分开输出，支持只重跑失败目标
退出码	`0` 表示成功，非 `0` 让 cron/GitLab 能发现失败
dry-run	高风险修改先展示将要操作的对象和动作

命令行入口¶

import argparse
import logging


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="批量检查服务健康状态")
    parser.add_argument("--input", required=True, help="目标清单")
    parser.add_argument("--output", required=True, help="结果文件")
    parser.add_argument("--timeout", type=float, default=5)
    parser.add_argument("--workers", type=int, default=5)
    parser.add_argument("--dry-run", action="store_true")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
    )
    logging.info("input=%s output=%s dry_run=%s", args.input, args.output, args.dry_run)
    return 0


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

使用 raise SystemExit(main()) 可以让调用方得到脚本返回码。异常不能只打印后仍返回 0。

API 批量检查¶

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests


def check_target(target: dict, timeout: float) -> dict:
    url = f"{target['base_url'].rstrip('/')}/actuator/health"
    try:
        response = requests.get(url, timeout=timeout)
        return {
            "name": target["name"],
            "url": url,
            "status_code": response.status_code,
            "ok": response.ok,
            "error": "",
        }
    except requests.RequestException as exc:
        return {
            "name": target["name"],
            "url": url,
            "status_code": None,
            "ok": False,
            "error": type(exc).__name__,
        }


def check_all(targets: list[dict], timeout: float, workers: int) -> list[dict]:
    results = []
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = [pool.submit(check_target, item, timeout) for item in targets]
        for future in as_completed(futures):
            results.append(future.result())
    return results

并发数要有上限。一次对几千台设备同时发请求可能压垮网络、目标服务或认证系统。先从 5～10 个 worker 开始，通过耗时和目标承载能力调整。

重试原则¶

适合重试：连接重置、短暂超时、HTTP 429 和部分 5xx。一般不应重试：认证失败、参数错误、权限不足和确定的业务拒绝。

from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential
import requests


@retry(
    retry=retry_if_exception_type((requests.Timeout, requests.ConnectionError)),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=8),
    reraise=True,
)
def get_json(url: str, timeout: float = 5) -> dict:
    response = requests.get(url, timeout=timeout)
    response.raise_for_status()
    return response.json()

修改类请求重试前必须确认接口具有幂等键或查询现有状态，否则“第一次成功但响应丢失”可能导致重复创建。

SSH 批量操作¶

SSH 自动化应优先执行查询命令。涉及修改时增加目标白名单、dry-run、审批和单机灰度。

from pathlib import Path
import paramiko


def run_command(host: str, username: str, key_file: Path, command: str) -> dict:
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.RejectPolicy())
    try:
        client.connect(
            hostname=host,
            username=username,
            key_filename=str(key_file),
            timeout=5,
            banner_timeout=5,
            auth_timeout=5,
        )
        _, stdout, stderr = client.exec_command(command, timeout=30)
        exit_code = stdout.channel.recv_exit_status()
        return {
            "host": host,
            "exit_code": exit_code,
            "stdout": stdout.read().decode(errors="replace"),
            "stderr": stderr.read().decode(errors="replace"),
        }
    finally:
        client.close()

不要自动接受未知 Host Key；先由受控流程把服务器指纹加入 known_hosts。日志中不要打印私钥路径之外的私钥内容、密码或完整 Token。

文件批处理¶

from pathlib import Path
import hashlib


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as file:
        for block in iter(lambda: file.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


for path in Path("input").glob("*.jar"):
    print(path.name, path.stat().st_size, sha256_file(path))

处理配置和制品时输出文件名、大小和校验值，便于确认不同服务器使用的是同一份文件。批量改文件时先写到临时文件，校验成功后再原子替换，并保留备份或版本控制。

子进程调用¶

import subprocess

result = subprocess.run(
    ["systemctl", "is-active", "myapp"],
    text=True,
    capture_output=True,
    timeout=10,
    check=False,
)

使用参数列表，不要把不可信输入拼成 shell=True 字符串。始终设置超时并检查 returncode、stdout 和 stderr。

定时运行¶

*/10 * * * * cd /opt/ops-automation && /usr/local/bin/uv run --locked python main.py --input config.yml --output output/latest.json >> /var/log/ops-automation.log 2>&1

长期服务或关键任务优先使用 systemd timer，让运行身份、资源限制、超时、状态和日志更清晰。定时任务要防止上一次未结束又启动下一次，可使用文件锁、数据库锁或任务平台的并发控制。

上线检查¶

在脱敏测试数据和测试主机上运行。
验证 dry-run 输出的目标数量和动作。
设置超时、并发上限、重试次数和总执行时限。
单个目标失败不应丢失整批结果；输出失败清单。
修改类任务先灰度一台，验证后再分批。
记录脚本 Git SHA、uv.lock、参数、执行用户和批次号。
明确回滚方式和停止开关。

常见问题¶

现象	优先检查
脚本一直不结束	HTTP、SSH、子进程是否都设置超时；线程池是否有卡住任务
定时任务能手工运行但 cron 失败	工作目录、完整命令路径、环境变量、权限和 uv 缓存
运行结果时好时坏	并发过高、下游限流、重试范围过宽、DNS 或网络抖动
部分主机被重复修改	是否具备幂等判断、批次状态和失败重跑机制
日志显示成功但实际失败	是否忽略返回码、HTTP 状态或 stderr