运维自动化是DevOps的核心。Python凭借丰富的库,可编写CI/CD脚本、管理Docker容器、调用Ansible等。本文将实战演示如何用Python打造自动化运维工具。
1. 使用Python操作Docker
安装docker库:
bash
复制
下载
pip install docker
列出容器
python
复制
下载
import docker client = docker.from_env() for container in client.containers.list(all=True): print(container.name, container.status)
构建镜像
python
复制
下载
image, logs = client.images.build(path="./app", tag="myapp:latest") for log in logs: print(log)
运行容器
python
复制
下载
container = client.containers.run( "myapp:latest", detach=True, ports={"8000/tcp": 8000}, environment={"ENV": "prod"} ) print(container.id)2. 编写CI/CD脚本
以GitLab CI为例,但用Python做本地预检。
检查代码风格
python
复制
下载
import subprocess import sys def run_check(cmd): result = subprocess.run(cmd, shell=True, capture_output=True, text=True) if result.returncode != 0: print(f"失败: {cmd}\n{result.stderr}") sys.exit(1) run_check("black --check .") run_check("flake8 .") run_check("pytest")自动版本发布
python
复制
下载
import os import re import subprocess def bump_version(part="patch"): with open("version.py") as f: content = f.read() version = re.search(r'__version__ = "(\d+)\.(\d+)\.(\d+)"', content) major, minor, patch = map(int, version.groups()) if part == "major": major += 1; minor = 0; patch = 0 elif part == "minor": minor += 1; patch = 0 else: patch += 1 new_version = f"{major}.{minor}.{patch}" content = re.sub(r'__version__ = ".*"', f'__version__ = "{new_version}"', content) with open("version.py", "w") as f: f.write(content) subprocess.run(["git", "tag", f"v{new_version}"]) subprocess.run(["git", "push", "--tags"]) print(f"版本更新至 {new_version}") bump_version("patch")3. 调用Ansible
安装ansible-runner:
bash
复制
下载
pip install ansible-runner
执行Playbook
python
复制
下载
import ansible_runner def run_playbook(playbook, inventory): r = ansible_runner.run( private_data_dir=".", playbook=playbook, inventory=inventory, quiet=False ) print(f"状态: {r.status}") print(f"输出: {r.stdout.read()}") return r.rc rc = run_playbook("deploy.yml", "hosts.ini") if rc != 0: raise SystemExit("部署失败")4. 自动化监控与告警
python
复制
下载
import psutil import requests import time def check_system(): cpu = psutil.cpu_percent(interval=1) mem = psutil.virtual_memory().percent disk = psutil.disk_usage("/").percent return cpu, mem, disk def send_alert(msg): webhook = "https://hooks.slack.com/services/XXX" requests.post(webhook, json={"text": msg}) while True: cpu, mem, disk = check_system() if cpu > 90 or mem > 90 or disk > 90: send_alert(f"资源告警: CPU={cpu}%, MEM={mem}%, DISK={disk}%") time.sleep(60)5. 集成到CI/CD流水线
GitHub Actions:使用Python脚本作为步骤。
Jenkins:在Pipeline中调用
python deploy.py。GitLab CI:
script: - python ci_check.py。
6. 最佳实践
使用
argparse或click构建CLI。日志用
logging模块。敏感信息用环境变量或Vault。
异常处理与重试。
单元测试脚本。
7. 实战:一键部署脚本
python
复制
下载
#!/usr/bin/env python3 import argparse import subprocess import docker import ansible_runner def build(): client = docker.from_env() client.images.build(path="./app", tag="myapp:latest") def deploy(): r = ansible_runner.run(playbook="deploy.yml", inventory="prod.ini") if r.rc != 0: raise SystemExit("部署失败") def main(): parser = argparse.ArgumentParser() parser.add_argument("action", choices=["build", "deploy", "all"]) args = parser.parse_args() if args.action in ("build", "all"): build() if args.action in ("deploy", "all"): deploy() if __name__ == "__main__": main()总结
Python在自动化运维中用途广泛,从容器管理到CI/CD,再到配置管理。结合Docker、Ansible、requests等库,可打造高效、可维护的运维工具。建议从简单脚本开始,逐步模块化。