周志洋

个人站

持续学习 才能不被淘汰


Python DevOps-CI/CD流水线详解

CI/CD流水线的重要性

CI/CD(持续集成/持续部署)是现代DevOps实践的核心,能够自动化代码测试、构建和部署流程,显著提升开发效率和代码质量。Python项目通过CI/CD流水线可以实现自动化测试、代码质量检查、容器化构建和自动化部署。本文将从GitHub Actions到Jenkins,全面介绍Python CI/CD流水线的最佳实践。

GitHub Actions

1. 基础CI/CD流水线

# .github/workflows/ci-cd.yml
name: Python CI/CD Pipeline

# 触发条件
on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]
  schedule:
    # 每天凌晨2点运行
    - cron: '0 2 * * *'

jobs:
  # 测试任务
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: [3.7, 3.8, 3.9]
    
    steps:
    # 检出代码
    - name: Checkout code
      uses: actions/checkout@v2
    
    # 设置Python环境
    - name: Set up Python $
      uses: actions/setup-python@v2
      with:
        python-version: $
        cache: 'pip'
    
    # 安装依赖
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install -r requirements-dev.txt
    
    # 代码质量检查
    - name: Lint with flake8
      run: |
        pip install flake8
        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
    
    # 类型检查
    - name: Type check with mypy
      run: |
        pip install mypy
        mypy . --ignore-missing-imports || true
    
    # 运行测试
    - name: Run tests
      run: |
        pytest --cov=./ --cov-report=xml --cov-report=html
    
    # 上传覆盖率报告
    - name: Upload coverage to Codecov
      uses: codecov/codecov-action@v2
      with:
        file: ./coverage.xml
        flags: unittests
        name: codecov-umbrella

2. 多环境部署流水线

# .github/workflows/deploy.yml
name: Deploy Pipeline

on:
  push:
    branches: [ main ]
  workflow_dispatch:
    inputs:
      environment:
        description: '部署环境'
        required: true
        default: 'staging'
        type: choice
        options:
          - staging
          - production

jobs:
  build:
    runs-on: ubuntu-latest
    
    steps:
    - name: Checkout code
      uses: actions/checkout@v2
    
    - name: Set up Docker Buildx
      uses: docker/setup-buildx-action@v1
    
    - name: Login to Docker Hub
      uses: docker/login-action@v1
      with:
        username: $
        password: $
    
    - name: Build and push Docker image
      uses: docker/build-push-action@v2
      with:
        context: .
        push: true
        tags: |
          myapp:$
          myapp:latest
        cache-from: type=registry,ref=myapp:buildcache
        cache-to: type=registry,ref=myapp:buildcache,mode=max
  
  deploy-staging:
    needs: build
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    
    steps:
    - name: Deploy to staging
      uses: appleboy/ssh-action@master
      with:
        host: $
        username: $
        key: $
        script: |
          docker pull myapp:$
          docker stop myapp-staging || true
          docker rm myapp-staging || true
          docker run -d --name myapp-staging -p 8000:8000 myapp:$
  
  deploy-production:
    needs: build
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main' && github.event_name == 'workflow_dispatch'
    environment: production
    
    steps:
    - name: Deploy to production
      uses: appleboy/ssh-action@master
      with:
        host: $
        username: $
        key: $
        script: |
          docker pull myapp:$
          docker stop myapp-prod || true
          docker rm myapp-prod || true
          docker run -d --name myapp-prod -p 8000:8000 myapp:$

3. Python项目CI/CD最佳实践

# 示例:pytest配置文件
# pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = 
    --cov=src
    --cov-report=html
    --cov-report=term-missing
    --cov-report=xml
    --strict-markers
    --verbose

# 示例:requirements-dev.txt
# 开发环境依赖
pytest>=6.0.0
pytest-cov>=2.10.0
pytest-mock>=3.3.0
flake8>=3.8.0
mypy>=0.800
black>=21.0.0
isort>=5.0.0
pre-commit>=2.10.0

# 示例:pre-commit配置
# .pre-commit-config.yaml
repos:
  - repo: https://github.com/psf/black
    rev: 21.6b0
    hooks:
      - id: black
        language_version: python3.9
  
  - repo: https://github.com/pycqa/isort
    rev: 5.9.3
    hooks:
      - id: isort
  
  - repo: https://github.com/pycqa/flake8
    rev: 3.9.2
    hooks:
      - id: flake8
        args: [--max-line-length=127]

Jenkins Pipeline

1. 声明式Pipeline

// Jenkinsfile
pipeline {
    agent any
    
    environment {
        PYTHON_VERSION = '3.9'
        DOCKER_IMAGE = 'myapp'
        DOCKER_REGISTRY = 'registry.example.com'
    }
    
    options {
        buildDiscarder(logRotator(numToKeepStr: '10'))
        timeout(time: 30, unit: 'MINUTES')
        timestamps()
    }
    
    stages {
        // 代码检出
        stage('Checkout') {
            steps {
                checkout scm
            }
        }
        
        // 环境准备
        stage('Setup') {
            steps {
                sh '''
                    python${PYTHON_VERSION} -m venv venv
                    . venv/bin/activate
                    pip install --upgrade pip
                    pip install -r requirements.txt
                    pip install -r requirements-dev.txt
                '''
            }
        }
        
        // 代码质量检查
        stage('Quality Check') {
            parallel {
                stage('Lint') {
                    steps {
                        sh '''
                            . venv/bin/activate
                            flake8 . --count --statistics
                        '''
                    }
                }
                stage('Type Check') {
                    steps {
                        sh '''
                            . venv/bin/activate
                            mypy . --ignore-missing-imports || true
                        '''
                    }
                }
            }
        }
        
        // 运行测试
        stage('Test') {
            steps {
                sh '''
                    . venv/bin/activate
                    pytest --cov=./ --cov-report=xml --cov-report=html --junitxml=test-results.xml
                '''
            }
            post {
                always {
                    publishTestResults testResultsPattern: 'test-results.xml'
                    publishHTML([
                        reportDir: 'htmlcov',
                        reportFiles: 'index.html',
                        reportName: 'Coverage Report'
                    ])
                }
            }
        }
        
        // 构建Docker镜像
        stage('Build') {
            steps {
                script {
                    def imageTag = "${DOCKER_REGISTRY}/${DOCKER_IMAGE}:${env.BUILD_NUMBER}"
                    def latestTag = "${DOCKER_REGISTRY}/${DOCKER_IMAGE}:latest"
                    
                    sh """
                        docker build -t ${imageTag} -t ${latestTag} .
                    """
                }
            }
        }
        
        // 安全扫描
        stage('Security Scan') {
            steps {
                sh '''
                    docker run --rm -v /var/run/docker.sock:/var/run/docker.sock \
                        aquasec/trivy image ${DOCKER_REGISTRY}/${DOCKER_IMAGE}:${env.BUILD_NUMBER}
                '''
            }
        }
        
        // 部署到测试环境
        stage('Deploy to Staging') {
            when {
                branch 'develop'
            }
            steps {
                sh '''
                    docker push ${DOCKER_REGISTRY}/${DOCKER_IMAGE}:${env.BUILD_NUMBER}
                    docker push ${DOCKER_REGISTRY}/${DOCKER_IMAGE}:latest
                    # 部署到staging环境
                    ssh user@staging-server "docker pull ${DOCKER_REGISTRY}/${DOCKER_IMAGE}:${env.BUILD_NUMBER} && docker-compose up -d"
                '''
            }
        }
        
        // 部署到生产环境
        stage('Deploy to Production') {
            when {
                branch 'main'
            }
            steps {
                input message: '确认部署到生产环境?', ok: 'Deploy'
                sh '''
                    docker push ${DOCKER_REGISTRY}/${DOCKER_IMAGE}:${env.BUILD_NUMBER}
                    docker push ${DOCKER_REGISTRY}/${DOCKER_IMAGE}:latest
                    # 部署到生产环境
                    ssh user@prod-server "docker pull ${DOCKER_REGISTRY}/${DOCKER_IMAGE}:${env.BUILD_NUMBER} && docker-compose up -d"
                '''
            }
        }
    }
    
    post {
        always {
            cleanWs()
        }
        success {
            emailext(
                subject: "构建成功: ${env.JOB_NAME} - ${env.BUILD_NUMBER}",
                body: "构建成功!",
                to: "${env.CHANGE_AUTHOR_EMAIL}"
            )
        }
        failure {
            emailext(
                subject: "构建失败: ${env.JOB_NAME} - ${env.BUILD_NUMBER}",
                body: "构建失败,请检查日志。",
                to: "${env.CHANGE_AUTHOR_EMAIL}"
            )
        }
    }
}

2. 脚本式Pipeline

// Jenkinsfile (Scripted)
node {
    def pythonVersion = '3.9'
    def dockerImage = 'myapp'
    def dockerRegistry = 'registry.example.com'
    
    stage('Checkout') {
        checkout scm
    }
    
    stage('Setup Environment') {
        sh """
            python${pythonVersion} -m venv venv
            . venv/bin/activate
            pip install --upgrade pip
            pip install -r requirements.txt
            pip install -r requirements-dev.txt
        """
    }
    
    stage('Test') {
        try {
            sh """
                . venv/bin/activate
                pytest --cov=./ --cov-report=xml --junitxml=test-results.xml
            """
            junit 'test-results.xml'
        } catch (Exception e) {
            currentBuild.result = 'FAILURE'
            throw e
        }
    }
    
    stage('Build Docker Image') {
        def imageTag = "${dockerRegistry}/${dockerImage}:${env.BUILD_NUMBER}"
        sh "docker build -t ${imageTag} ."
        sh "docker push ${imageTag}"
    }
    
    stage('Deploy') {
        if (env.BRANCH_NAME == 'main') {
            sh """
                ssh user@prod-server "docker pull ${imageTag} && docker-compose up -d"
            """
        }
    }
}

CI/CD最佳实践

1. 测试策略

# 示例:分层测试结构
# tests/
#   ├── unit/          # 单元测试
#   │   ├── test_models.py
      └── test_utils.py
#   ├── integration/   # 集成测试
#   │   ├── test_api.py
#   │   └── test_database.py
#   └── e2e/           # 端到端测试
#       └── test_user_flow.py

# 示例:单元测试
# tests/unit/test_models.py
import pytest
from datetime import datetime
from src.models import User

class TestUser:
    def test_user_creation(self):
        """测试用户创建"""
        user = User(name="张三", email="zhangsan@example.com")
        assert user.name == "张三"
        assert user.email == "zhangsan@example.com"
        assert user.created_at is not None
    
    def test_user_validation(self):
        """测试用户验证"""
        with pytest.raises(ValueError):
            User(name="", email="invalid")

# 示例:集成测试
# tests/integration/test_api.py
import pytest
from fastapi.testclient import TestClient
from src.main import app

client = TestClient(app)

def test_create_user():
    """测试创建用户API"""
    response = client.post(
        "/api/users",
        json={"name": "李四", "email": "lisi@example.com"}
    )
    assert response.status_code == 201
    assert response.json()["name"] == "李四"

2. 部署策略

# 示例:蓝绿部署脚本
# scripts/deploy.py
#!/usr/bin/env python3
"""蓝绿部署脚本"""
import subprocess
import sys
from datetime import datetime

def deploy_blue_green(environment='staging'):
    """执行蓝绿部署"""
    timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
    blue_service = f"myapp-blue-{timestamp}"
    green_service = f"myapp-green-{timestamp}"
    
    print(f"开始蓝绿部署到 {environment} 环境...")
    
    # 1. 部署新版本(蓝环境)
    print(f"部署蓝环境: {blue_service}")
    subprocess.run([
        "docker", "run", "-d",
        "--name", blue_service,
        "-p", "8001:8000",
        "myapp:latest"
    ], check=True)
    
    # 2. 健康检查
    print("执行健康检查...")
    health_check = subprocess.run([
        "curl", "-f", "http://localhost:8001/health"
    ], capture_output=True)
    
    if health_check.returncode != 0:
        print("健康检查失败,回滚...")
        subprocess.run(["docker", "stop", blue_service])
        subprocess.run(["docker", "rm", blue_service])
        sys.exit(1)
    
    # 3. 切换流量到新版本
    print("切换流量到新版本...")
    # 更新负载均衡器配置
    
    # 4. 停止旧版本(绿环境)
    print("停止旧版本...")
    # 停止旧容器
    
    print("部署完成!")

if __name__ == "__main__":
    environment = sys.argv[1] if len(sys.argv) > 1 else 'staging'
    deploy_blue_green(environment)

总结

CI/CD流水线的关键要点:

  1. 自动化测试:单元测试、集成测试、端到端测试全覆盖
  2. 代码质量:Lint检查、类型检查、代码覆盖率
  3. 容器化构建:Docker镜像构建和推送
  4. 多环境部署:开发、测试、生产环境分离
  5. 安全扫描:依赖漏洞扫描、镜像安全扫描
  6. 部署策略:蓝绿部署、滚动更新、金丝雀发布
  7. 监控告警:构建状态通知、部署结果通知

掌握这些CI/CD技能,可以建立高效的自动化开发流程,显著提升开发效率和代码质量,为Python项目提供强大的持续集成和持续部署支持。


转载请注明:周志洋的博客 » Python DevOps-CI/CD流水线详解

打赏一个呗

取消

感谢您的支持,我会继续努力的!

扫码支持
扫码支持
扫码打赏,你说多少就多少

打开支付宝扫一扫,即可进行扫码打赏哦