Python DevOps与自动化

DevOps强调开发和运维协作，Python是自动化运维的主流语言。

CI/CD持续集成部署

CI/CD概念

CI_CD_CONCEPTS = {
    'CI': '持续集成，自动构建、测试',
    'CD': '持续部署，自动部署到环境',
    'Pipeline': '构建、测试、部署的自动化流程',
    'Artifact': '构建产物，可部署的包'
}

# CI/CD流程：
# Code → Build → Test → Deploy → Monitor

GitLab CI配置

# .gitlab-ci.yml

stages:
  - build
  - test
  - deploy

build:
  stage: build
  script:
    - pip install -r requirements.txt
    - python setup.py build
  artifacts:
    paths:
      - dist/

test:
  stage: test
  script:
    - pip install pytest pytest-cov
    - pytest --cov=app tests/
  coverage: '/TOTAL.*\s+(\d+%)/'

deploy:
  stage: deploy
  script:
    - pip install twine
    - twine upload dist/*
  only:
    - main

GitHub Actions配置

# .github/workflows/ci.yml

name: CI Pipeline

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  build:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-cov          

      - name: Run tests
        run: pytest --cov=app tests/

      - name: Upload coverage
        uses: codecov/codecov-action@v3

Docker容器化

Dockerfile编写

# Dockerfile

# 基础镜像
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 设置环境变量
ENV PYTHONUNBUFFERED=1
ENV APP_ENV=production

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["python", "app.py"]

Docker Compose多服务

# docker-compose.yml

version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgres://db:5432
    depends_on:
      - db
      - redis

  db:
    image: postgres:15
    environment:
      POSTGRES_PASSWORD: secret
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7

volumes:
  pgdata:

Docker Python SDK

import docker

client = docker.from_env()

# 构建镜像
image = client.images.build(path='.', tag='myapp:latest')

# 运行容器
container = client.containers.run(
    'myapp:latest',
    ports={'8000/tcp': 8000},
    detach=True
)

# 查看容器
containers = client.containers.list()

# 停止容器
container.stop()
container.remove()

# 执行命令
output = container.exec_run('python -c "print(1+1)"')

自动化脚本

部署自动化

import subprocess
import os

class Deployer:
    "部署自动化"

    def deploy(self, environment):
        steps = [
            self.pull_code,
            self.install_dependencies,
            self.run_tests,
            self.build_package,
            self.deploy_to_env
        ]

        for step in steps:
            result = step(environment)
            if not result:
                raise Exception(f"Deploy failed at {step.__name__}")

    def pull_code(self, env):
        subprocess.run(['git', 'pull'], check=True)
        return True

    def install_dependencies(self, env):
        subprocess.run(['pip', 'install', '-r', 'requirements.txt'], check=True)
        return True

    def run_tests(self, env):
        result = subprocess.run(['pytest', 'tests/'], capture_output=True)
        return result.returncode == 0

    def build_package(self, env):
        subprocess.run(['python', 'setup.py', 'build'], check=True)
        return True

    def deploy_to_env(self, env):
        if env == 'production':
            subprocess.run(['docker', 'compose', 'up', '-d'], check=True)
        return True

监控脚本

import psutil
import time
import requests

class Monitor:
    "系统监控"

    def check_health(self):
        metrics = {
            'cpu': psutil.cpu_percent(),
            'memory': psutil.virtual_memory().percent,
            'disk': psutil.disk_usage('/').percent,
            'network': self.check_network()
        }

        alerts = []
        if metrics['cpu'] > 80:
            alerts.append('CPU usage high')
        if metrics['memory'] > 90:
            alerts.append('Memory usage critical')

        return metrics, alerts

    def check_network(self):
        try:
            requests.get('https://api.example.com/health', timeout=5)
            return 'healthy'
        except:
            return 'unhealthy'

    def log_metrics(self):
        metrics, alerts = self.check_health()
        # 发送到监控系统
        requests.post('https://monitor.example.com/api/metrics', json=metrics)

        if alerts:
            self.send_alert(alerts)

    def send_alert(self, alerts):
        # 发送告警
        requests.post('https://alert.example.com/api/send', json={'alerts': alerts})

配置管理

Ansible自动化

# playbook.yml

- name: Deploy Python Application
  hosts: webservers
  become: yes

  tasks:
    - name: Install Python
      apt:
        name: python3.11
        state: present

    - name: Install dependencies
      pip:
        requirements: /app/requirements.txt

    - name: Copy application
      copy:
        src: ./app/
        dest: /app/

    - name: Start service
      systemd:
        name: myapp
        state: started
        enabled: yes

Python配置管理

from pydantic import BaseSettings

class Settings(BaseSettings):
    "环境配置"
    app_name: str = 'myapp'
    debug: bool = False

    # 数据库配置
    db_host: str
    db_port: int = 5432
    db_name: str

    # Redis配置
    redis_host: str = 'localhost'

    class Config:
        env_file = '.env'

# 环境区分
class DevelopmentSettings(Settings):
    debug = True

class ProductionSettings(Settings):
    debug = False

def get_settings():
    env = os.getenv('APP_ENV', 'development')
    if env == 'production':
        return ProductionSettings()
    return DevelopmentSettings()

日志管理

结构化日志

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_obj = {
            'timestamp': datetime.utcnow().isoformat(),
            'level': record.levelname,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno
        }
        return json.dumps(log_obj)

# 配置日志
logger = logging.getLogger()
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

# 使用
logger.info('Application started')
logger.error('Database connection failed', extra={'db_host': 'localhost'})

ELK日志收集

import logging
from logstash_async.handler import AsynchronousLogstashHandler

# 发送日志到Logstash
handler = AsynchronousLogstashHandler(
    host='logstash.example.com',
    port=5959,
    database_path='logstash.db'
)

logger = logging.getLogger()
logger.addHandler(handler)

# 结构化日志
logger.info('User logged in', extra={
    'user_id': 123,
    'ip': '192.168.1.1'
})

常用DevOps工具

注意：DevOps工具链要根据团队规模和技术能力选择，避免过度复杂化。

要点总结

• CI/CD：GitLab CI/GitHub Actions自动化构建、测试、部署流程
• Docker容器化：Dockerfile定义镜像、Compose编排多服务
• 自动化脚本：部署流程脚本化、监控指标采集、告警发送
• 配置管理：Ansible批量配置、环境变量分离、Settings类管理
• 日志管理：结构化JSON日志、ELK收集分析、集中管理

存放路径：articles/PYTHON/专家/生态与工具链/DevOps与自动化.md