Python 运维入门

摘要：Shell 适合组合命令，Python 适合处理逻辑。本文给出运维脚本最常用的三块：用 requests 调 HTTP 接口、用 paramiko 做 SSH 远程执行、用 subprocess 调本地命令，并强调超时、异常与凭证处理。

适用环境

python3 --version
python3 -m venv ~/venv-ops && source ~/venv-ops/bin/activate
pip install requests paramiko
python3 -c "import requests, paramiko; print('ok')"

操作步骤

一、用 requests 调接口（务必设超时）

import os
import requests

BASE = os.environ.get("API_BASE", "https://api.example.com")

def get_health(timeout: float = 5.0) -> dict:
    """拉取健康检查接口，失败抛异常"""
    r = requests.get(f"{BASE}/v1/health", timeout=timeout)
    r.raise_for_status()          # 4xx/5xx 直接抛错
    return r.json()

def post_with_retry(path: str, payload: dict, retries: int = 3) -> dict:
    """带重试的 POST，只对网络类异常重试"""
    last = None
    for i in range(retries):
        try:
            r = requests.post(f"{BASE}{path}", json=payload, timeout=(3.0, 10.0))
            r.raise_for_status()
            return r.json()
        except requests.exceptions.RequestException as e:
            last = e
    raise RuntimeError(f"请求失败，已重试 {retries} 次：{last}")

if __name__ == "__main__":
    print(get_health())

超时参数 (3.0, 10.0) 分别表示「连接超时」和「读取超时」，两者都应设置。

注意

requests 默认不限超时。一个卡死的接口会让脚本永久挂起，在 crontab 里表现为多个进程越积越多。任何网络调用都必须显式传 timeout。

二、用 subprocess 调本地命令

import subprocess

def run(cmd: list[str], timeout: int = 60) -> tuple[int, str, str]:
    """执行命令，返回 (退出码, stdout, stderr)，不抛异常"""
    p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    return p.returncode, p.stdout.strip(), p.stderr.strip()

rc, out, err = run(["systemctl", "is-active", "nginx"])
print(rc, out)

rc, out, err = run(["df", "-P", "/"])
usage = out.splitlines()[1].split()[4].rstrip("%")
print("根分区使用率:", usage)

三、用 paramiko 做 SSH 远程执行

import paramiko

def ssh_run(host: str, user: str, key_path: str, cmd: str, timeout: int = 30):
    """密钥方式远程执行命令，返回 (rc, stdout, stderr)"""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(
            hostname=host, username=user, key_filename=key_path,
            timeout=timeout, banner_timeout=timeout,
        )
        stdin, stdout, stderr = client.exec_command(cmd, timeout=timeout)
        rc = stdout.channel.recv_exit_status()
        return rc, stdout.read().decode().strip(), stderr.read().decode().strip()
    finally:
        client.close()

rc, out, err = ssh_run("10.0.1.10", "ops", "/home/ops/.ssh/id_ed25519", "uptime")
print(rc, out)

危险

AutoAddPolicy 会自动接受未知主机密钥，存在中间人攻击风险。生产环境应改用 RejectPolicy 并预先维护 known_hosts，或用 client.load_system_host_keys()。

四、凭证永远不写进代码

import os

# 从环境变量读取，缺失时立即报错而不是用默认值硬跑
def require_env(name: str) -> str:
    val = os.environ.get(name)
    if not val:
        raise SystemExit(f"缺少环境变量 {name}，请先在运行时注入")
    return val

TOKEN = require_env("API_TOKEN")
KEY   = require_env("SSH_KEY_PATH")

# 运行前注入
export API_TOKEN='xxxx'
export SSH_KEY_PATH="$HOME/.ssh/id_ed25519"
python3 check.py

五、脚本工程结构

#!/usr/bin/env python3
"""巡检示例：检查本机磁盘与服务状态，输出 JSON。"""
import argparse
import json
import logging
import sys
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    stream=sys.stderr,
)
log = logging.getLogger("ops")

def parse_args():
    p = argparse.ArgumentParser(description="主机巡检")
    p.add_argument("--warn", type=int, default=80, help="磁盘告警阈值")
    p.add_argument("--json", action="store_true", help="输出 JSON")
    return p.parse_args()

def main() -> int:
    args = parse_args()
    log.info("开始巡检 warn=%s", args.warn)
    result = {"time": datetime.now().isoformat(timespec="seconds")}
    # ... 采集逻辑 ...
    if args.json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        for k, v in result.items():
            print(f"{k}: {v}")
    return 0

if __name__ == "__main__":
    sys.exit(main())

main() 返回退出码，配合 sys.exit() 让脚本能被 crontab 和监控系统正确判断成败。

六、依赖与运行

# 固定依赖版本，避免环境不一致
pip freeze > requirements.txt
cat requirements.txt
# 新环境一键还原
pip install -r requirements.txt

# 给脚本加执行权限
chmod +x check.py && ./check.py --warn 85

验证

• [ ] requests 调用均带 timeout 参数，raise_for_status() 已使用
• [ ] subprocess 用列表形式传参（无 shell=True），且带超时
• [ ] paramiko 使用密钥登录，凭证从环境变量读取
• [ ] 脚本 main() 返回退出码，可用 echo $? 检查
• [ ] requirements.txt 存在且能在干净虚拟环境还原

常见坑

• 用 shell=True 拼接字符串：参数含空格或分号时会被当成命令执行，是命令注入的经典入口。
• 不设超时：脚本永久挂起，cron 任务叠加导致机器负载异常。
• 把密码写进脚本并提交 Git：应改用环境变量或密钥管理服务。
• 捕获所有异常却不处理：except Exception: pass 会掩盖真实故障，至少记录日志。
• print 当日志用：没有时间戳和级别，排查时无法还原顺序，应使用 logging。

参考资料

• Requests 官方文档
• Paramiko 官方文档
• Python subprocess 文档
• Python logging 指南