Python实现Linux系统上CI/CD工作流的方法详解

目录

• 完整实现代码
• 使用说明
• 扩展功能建议
• 关键注意事项

完整实现代码

#!/usr/bin/env python3
import subprocess
import os
import sys
import time
import logging
import argparse
import shlex
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("ci_cd.log"), logging.StreamHandler()]
)
logger = logging.getLogger("CI/CD")

class CICD:
    def __init__(self, repo_path, branch="main"):
        self.repo_path = Path(repo_path).resolve()
        self.branch = branch
        self.last_commit = None
        self._check_dependencies(["git", "docker"])
        
    def _check_dependencies(self, deps):
        for cmd in deps:
            try:
                subprocess.run([cmd, "--version"], check=True, 
                              stdout=subprocess.DEVNULL)
            except Exception:
                raise RuntimeError(f"缺少依赖: {cmd}")

    def check_updates(self):
        try:
            # 检查远程更新
            subprocess.run(["git", "-C", str(self.repo_path), "fetch"], check=True)
            
            # 比较本地与远程差异
            diff = subprocess.run(
                ["git", "-C", str(self.repo_path), "diff", "--shortstat", f"origin/{self.branch}"],
                stdout=subprocess.PIPE
            )
            if not diff.stdout.strip():
                return False

            # 拉取最新代码
            subprocess.run(
                ["git", "-C", str(self.repo_path), "pull", "origin", self.branch],
                check=True
            )
            
            # 获取最新提交ID
            new_commit = subprocess.check_output(
                ["git", "-C", str(self.repo_path), "rev-parse", "HEAD"]
            ).decode().strip()
            
            if new_commit != self.last_commit:
                self.last_commit = new_commit
                return True
            return False
            
        except subprocess.CalledProcessError as e:
            logger.error(f"代码更新失败: {str(e)}")
            raise

    def run_tests(self):
        try:
            subprocess.run(["pytest", "tests/"], cwd=self.repo_path, check=True)
            return True
        except subprocess.CalledProcessError:
            logger.error("测试失败")
            return False

    def build(self):
        try:
            subprocess.run(
                ["docker", "build", "-t", "myapp:latest", "."],
                cwd=self.repo_path, check=True
            )
            return True
        except subprocess.CalledProcessError:
            logger.error("构建失败")
            return False

    def deploy(self):
        try:
            # 停止旧容器
            subprocess.run(["docker", "stop", "myapp"], check=False)
            # 启动新容器
 php           subprocess.run(
                ["docker", "run", "--rm", "-d", "--name", "myapp", "-p", "8000:8000", "myapp:latest"],
                check=True
            )
            return True
        except subprocess.CalledProcessError:
            logger.error("部署失败")
            return False

    def run_pipeline(self):
        if not self.check_updates():
            logger.info("无代码更新")
            return True
            
        logger.info(f"检测到新提交: {self.last_commit[:8]}")
        return self.run_tests() andpython self.build() and self.deploy()

def main():
    parser = argparse.ArgumentParser(description="CI/CD 流水线")
    parser.add_argument("--repo", required=True, help="仓库路径")
    parser.add_argument("--branch", default="main", help="监控分支")
    parser.add_argument("--daemon", action="store_true", help="守护模式")
    parser.add_argument("--interval", type=int, default=60, help="检查间隔")
    
    args = parser.parse_args()
    
    try:
        ci = CICD(args.repo, args.branch)
        if args.daemon:
            while True:
                ci.run_pipeline()
                time.sleep(args.interval)
        else:
            success = ci.run_pipeline()
            sys.exit(0 if success else 1)
    except Exception as 编程e:
        logger.error(f"流程异常: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()

使用说明

安装依赖：

pip install pytest docker

运行方式：

# 单次运行模式
python ci_cd.py --repo /path/to/repo --branch main

# 守护进程模式（每5分钟检查一次）
python ci_cd.py --repo ~/myapp --daemon --interval 300

自定义命令参数：

# 使用自定义测试命令
python ci_cd.py --repo ~/project \
  --test-cmd "npm run test" \
  --build-cmd "docker build -t myapp:v1 ." \
  --deploy-cmd "kubectl apply -f deploy.yaml"

Git Hook 集成（可选）： 在 .git/hooks/post-receive 中添加：

#!/bin/sh
python /path/to/ci_cd.py --repo $(pwd)

扩展功能建议

通知功能：

# 添加至 CICD 类
def send_notification(self, message):
    import requests
    requests.post(
        "https://api.alert.com/notify",
        json={"text": f"[CI/CD] {message}"}
    )

# 在部署成功后调用
self.send_notification(f"部署成功: {self.last_commit[:8]}")

健康检查：

def health_check(self):
    import requests
    try:
        resp = requests.get("http://localhost:8000/health", timeout=5)
        return resp.status_code == 200
    except Exception:
        return False

配置文件支持： 创建 config.yaml：

repo: ~/mhttp://www.devze.comyapp
branch: dev
test_cmd: "npm test"
build_cmd: "docker build -t myapp:latest ."

关键注意事项

1.权限管理：

确保运行用户具有 Docker 执行权限

建议将用户加入 docker 用户组：

sudo usermod -aG docker $USER

2.安全建议：

• 通过环境变量管理敏感信息（如 API 密钥）
• 使用 --env-file 传递 Docker 环境变量

3.错误处理：

• 重要操作建议添加重试机制
• 部署失败时可自动回滚到上一个版本

4.性能优化：

• 使用 Docker 缓存加速构建
• 并行执行测试任务

到此这篇关于Python实现linux系统上CI/CD工作流的方法详解的文编程客栈章就介绍到这了,更多相关Python实现CI/CD工作流内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！