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-receive
Added 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 userdocker
User Group:
sudo usermod -aG docker $USER
2. Safety advice:
- Manage sensitive information (such as API keys) through environment variables
- use
--env-file
Passing 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!