Python實(shí)現(xiàn)Linux系統(tǒng)上CI/CD工作流的方法詳解

更新時(shí)間：2025年04月16日 09:03:19 作者：ak啊

在現(xiàn)代軟件開發(fā)中,持續(xù)集成（CI）和持續(xù)部署（CD）是提高開發(fā)效率、保證代碼質(zhì)量的重要手段,下面我們來(lái)看看如何使用Python實(shí)現(xiàn)Linux系統(tǒng)的CI/CD工作流吧

完整實(shí)現(xiàn)代碼

#!/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:
            # 檢查遠(yuǎn)程更新
            subprocess.run(["git", "-C", str(self.repo_path), "fetch"], check=True)
            
            # 比較本地與遠(yuǎn)程差異
            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("測(cè)試失敗")
            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("構(gòu)建失敗")
            return False

    def deploy(self):
        try:
            # 停止舊容器
            subprocess.run(["docker", "stop", "myapp"], check=False)
            # 啟動(dòng)新容器
            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("無(wú)代碼更新")
            return True
            
        logger.info(f"檢測(cè)到新提交: {self.last_commit[:8]}")
        return self.run_tests() and self.build() and self.deploy()

def main():
    parser = argparse.ArgumentParser(description="CI/CD 流水線")
    parser.add_argument("--repo", required=True, help="倉(cāng)庫(kù)路徑")
    parser.add_argument("--branch", default="main", help="監(jiān)控分支")
    parser.add_argument("--daemon", action="store_true", help="守護(hù)模式")
    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()

使用說(shuō)明

安裝依賴：

pip install pytest docker

運(yùn)行方式：

# 單次運(yùn)行模式
python ci_cd.py --repo /path/to/repo --branch main

# 守護(hù)進(jìn)程模式（每5分鐘檢查一次）
python ci_cd.py --repo ~/myapp --daemon --interval 300

自定義命令參數(shù)：

# 使用自定義測(cè)試命令
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)

擴(kuò)展功能建議

通知功能：

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

# 在部署成功后調(diào)用
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

配置文件支持：創(chuàng)建 config.yaml：

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

關(guān)鍵注意事項(xiàng)

1.權(quán)限管理：

確保運(yùn)行用戶具有 Docker 執(zhí)行權(quán)限

建議將用戶加入 docker 用戶組：

sudo usermod -aG docker $USER

2.安全建議：

通過環(huán)境變量管理敏感信息（如 API 密鑰）
使用 --env-file 傳遞 Docker 環(huán)境變量

3.錯(cuò)誤處理：

重要操作建議添加重試機(jī)制
部署失敗時(shí)可自動(dòng)回滾到上一個(gè)版本

4.性能優(yōu)化：

使用 Docker 緩存加速構(gòu)建
并行執(zhí)行測(cè)試任務(wù)

到此這篇關(guān)于Python實(shí)現(xiàn)Linux系統(tǒng)上CI/CD工作流的方法詳解的文章就介紹到這了,更多相關(guān)Python實(shí)現(xiàn)CI/CD工作流內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: