SoFunction
Updated on 2025-04-16

Detailed explanation of Python's method of implementing CI/CD workflow on Linux system

Complete implementation code

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

(
    level=,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[("ci_cd.log"), ()]
)
logger = ("CI/CD")

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

    def check_updates(self):
        try:
            # Check for remote updates            (["git", "-C", str(self.repo_path), "fetch"], check=True)
            
            # Comparison of local vs. remote differences            diff = (
                ["git", "-C", str(self.repo_path), "diff", "--shortstat", f"origin/{}"],
                stdout=
            )
            if not ():
                return False

            # Pull the latest code            (
                ["git", "-C", str(self.repo_path), "pull", "origin", ],
                check=True
            )
            
            # Get the latest submission 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  as e:
            (f"Code update failed: {str(e)}")
            raise

    def run_tests(self):
        try:
            (["pytest", "tests/"], cwd=self.repo_path, check=True)
            return True
        except :
            ("Test failed")
            return False

    def build(self):
        try:
            (
                ["docker", "build", "-t", "myapp:latest", "."],
                cwd=self.repo_path, check=True
            )
            return True
        except :
            ("Build failed")
            return False

    def deploy(self):
        try:
            # Stop old containers            (["docker", "stop", "myapp"], check=False)
            # Start a new container            (
                ["docker", "run", "--rm", "-d", "--name", "myapp", "-p", "8000:8000", "myapp:latest"],
                check=True
            )
            return True
        except :
            ("Deployment failed")
            return False

    def run_pipeline(self):
        if not self.check_updates():
            ("No code update")
            return True
            
        (f"New submission detected: {self.last_commit[:8]}")
        return self.run_tests() and () and ()

def main():
    parser = (description="CI/CD pipeline")
    parser.add_argument("--repo", required=True, help="Warehouse Path")
    parser.add_argument("--branch", default="main", help="Monitoring Branch")
    parser.add_argument("--daemon", action="store_true", help="Defense Mode")
    parser.add_argument("--interval", type=int, default=60, help="Check interval")
    
    args = parser.parse_args()
    
    try:
        ci = CICD(, )
        if :
            while True:
                ci.run_pipeline()
                ()
        else:
            success = ci.run_pipeline()
            (0 if success else 1)
    except Exception as e:
        (f"Process exception: {str(e)}")
        (1)

if __name__ == "__main__":
    main()

Instructions for use

Installation dependencies

pip install pytest docker

How to run

# Single run modepython ci_cd.py --repo /path/to/repo --branch main

# Daemon mode (check every 5 minutes)python ci_cd.py --repo ~/myapp --daemon --interval 300

Custom command parameters

# Use custom test commandspython ci_cd.py --repo ~/project \
  --test-cmd "npm run test" \
  --build-cmd "docker build -t myapp:v1 ." \
  --deploy-cmd "kubectl apply -f "

Git Hook Integration(Optional):.git/hooks/post-receiveAdded in:

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

Extended feature suggestions

Notification function

# Add to CICD classdef send_notification(self, message):
    import requests
    (
        "/notify",
        json={"text": f"[CI/CD] {message}"}
    )

# Call after deployment is successfulself.send_notification(f"Deployment successfully: {self.last_commit[:8]}")

Health check

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

Configuration file support: Create

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

Key considerations

1. Permission Management

Ensure that the running user has Docker execution permissions

It is recommended to join the userdockerUser Group:

sudo usermod -aG docker $USER

2. Safety advice

  • Manage sensitive information (such as API keys) through environment variables
  • use--env-filePassing Docker environment variables

3. Error handling

  • Important operations suggest adding a retry mechanism
  • Automatically rollback to the previous version when the deployment fails

4. Performance optimization

  • Speed ​​up building with Docker cache
  • Perform test tasks in parallel

This is the article about this detailed explanation of Python's method of implementing CI/CD workflows on Linux systems. For more related content on Python's CI/CD workflows, please search for my previous articles or continue browsing the following related articles. I hope everyone will support me in the future!