Python 全栈自动化从入门到实战：构建企业级自动化生态

pip install pytest pytest-cov

# math_utils.py
def add(a, b):
    return a + b

# test_math_utils.py
import pytest
from math_utils import add

def test_add_positive():
    assert add(2, 3) == 5

def test_add_negative():
    assert add(-1, -2) == -3

# 使用fixture实现前置/后置
@pytest.fixture
def sample_data():
    return [1, 2, 3]

def test_sum(sample_data):
    assert sum(sample_data) == 6

pytest --cov=. --cov-report=html

import requests
import pytest

BASE_URL = "https://api.example.com/v1"

def test_get_user():
    resp = requests.get(f"{BASE_URL}/users/1")
    assert resp.status_code == 200
    data = resp.json()
    assert data['id'] == 1
    assert 'name' in data

@pytest.mark.parametrize("user_id,expected_name", [
    (1, "Alice"),
    (2, "Bob"),
])
def test_multiple_users(user_id, expected_name):
    resp = requests.get(f"{BASE_URL}/users/{user_id}")
    assert resp.json()['name'] == expected_name

pytest --alluredir=./allure-results
allure serve ./allure-results

from playwright.sync_api import sync_playwright

def test_login():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        page = browser.new_page()
        page.goto("https://example.com/login")
        page.fill("#username", "admin")
        page.fill("#password", "secret")
        page.click("button[type=submit]")
        # 等待导航完成并验证跳转
        page.wait_for_url("https://example.com/dashboard")
        assert page.is_visible("h1:has-text('Dashboard')")
        browser.close()

pip install pre-commit
# 创建 .pre-commit-config.yaml
repos:
  - repo: https://github.com/psf/black
    rev: 23.1.0
    hooks:
      - id: black
  - repo: https://github.com/pycqa/flake8
    rev: 6.0.0
    hooks:
      - id: flake8
  - repo: https://github.com/python/mypy
    rev: v1.0.0
    hooks:
      - id: mypy

poetry new myproject
poetry add requests pytest --dev
poetry build   # 生成 wheel 和 sdist

FROM python:3.11-slim AS builder
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.11-slim
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY . /app
WORKDIR /app
CMD ["python", "app.py"]

name: CI/CD Pipeline
on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: |
          pip install poetry
          poetry install
      - name: Lint with flake8
        run: poetry run flake8 .
      - name: Test with pytest
        run: poetry run pytest --cov .
      - name: Build and push Docker image
        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
        run: |
          docker build -t myapp:${{ github.sha }} .
          docker tag myapp:${{ github.sha }} myregistry/myapp:latest
          docker push myregistry/myapp:latest
      - name: Deploy to Kubernetes
        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
        run: |
          kubectl set image deployment/myapp myapp=myregistry/myapp:${{ github.sha }} --record
          kubectl rollout status deployment/myapp

from kubernetes import client, config

config.load_kube_config()
apps_v1 = client.AppsV1Api()
# 更新 Deployment 镜像
apps_v1.patch_namespaced_deployment(
    name="myapp",
    namespace="default",
    body={"spec": {"template": {"spec": {"containers": [{"name": "myapp", "image": "myregistry/myapp:latest"}]}}}}
)
# 检查状态
apps_v1.read_namespaced_deployment_status(name="myapp", namespace="default")

from prometheus_client import start_http_server, Counter, Gauge
import random
import time

REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests')
ACTIVE_USERS = Gauge('active_users', 'Number of active users')

start_http_server(8000)
while True:
    REQUEST_COUNT.inc(random.randint(0, 10))
    ACTIVE_USERS.set(random.randint(0, 100))
    time.sleep(5)

import logging
from pythonjsonlogger import jsonlogger

logger = logging.getLogger()
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.info("User action", extra={"user_id": 123, "action": "login"})

import requests
import subprocess
from prometheus_api_client import PrometheusConnect

prom = PrometheusConnect(url="http://prometheus:9090")
result = prom.custom_query('rate(http_requests_total{status="500"}[5m])')
if result and float(result[0]['value'][1]) > 0.01:
    # 重启 Kubernetes deployment
    subprocess.run(["kubectl", "rollout", "restart", "deployment/myapp"])
    # 或扩容
    # subprocess.run(["kubectl", "scale", "deployment/myapp", "--replicas=5"])

import scrapy

class ExampleSpider(scrapy.Spider):
    name = "example"
    start_urls = ['https://example.com/products']

    def parse(self, response):
        for product in response.css('.product'):
            yield {
                'name': product.css('.name::text').get(),
                'price': product.css('.price::text').get()
            }
        next_page = response.css('a.next::attr(href)').get()
        if next_page:
            yield response.follow(next_page, self.parse)

import pandas as pd
from sqlalchemy import create_engine

df = pd.read_csv('raw_data.csv')
df.drop_duplicates(inplace=True)
df['price'] = df['price'].str.replace('$', '').astype(float)
engine = create_engine('postgresql://user:pass@host/db')
df.to_sql('products', engine, if_exists='append', index=False)

from apscheduler.schedulers.blocking import BlockingScheduler

def job():
    print("Running data sync...")
    # 执行爬虫或 ETL

scheduler = BlockingScheduler()
scheduler.add_job(job, 'cron', hour=2, minute=0)  # 每日凌晨2点
scheduler.start()

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_report(subject, body, to_emails):
    msg = MIMEMultipart()
    msg['From'] = 'automation@example.com'
    msg['To'] = ', '.join(to_emails)
    msg['Subject'] = subject
    msg.attach(MIMEText(body, 'html'))
    with smtplib.SMTP('smtp.example.com', 587) as server:
        server.starttls()
        server.login('user', 'pass')
        server.send_message(msg)

import requests
def send_dingtalk(message):
    webhook = 'https://oapi.dingtalk.com/robot/send?access_token=xxx'
    data = {"msgtype": "text", "text": {"content": message}}
    requests.post(webhook, json=data)

import streamlit as st
import pandas as pd

st.title("自动化运维看板")
df = pd.read_csv('metrics.csv')
st.line_chart(df)

from prometheus_api_client import PrometheusConnect
import time
import subprocess

def canary_analysis(namespace, deployment, canary_percent):
    prom = PrometheusConnect()
    # 假设已有 Service 分割流量，通过 istio 或 label selector
    # 此处简化：检查错误率
    query = f'sum(rate(http_requests_total{{namespace="{namespace}", deployment="{deployment}", status="500"}}[5m])) / sum(rate(http_requests_total{{namespace="{namespace}", deployment="{deployment}"}}[5m]))'
    result = prom.custom_query(query)
    error_rate = float(result[0]['value'][1]) if result else 0
    if error_rate > 0.01:
        subprocess.run(["kubectl", "rollout", "undo", f"deployment/{deployment}", "-n", namespace])
        send_dingtalk("生产异常，已自动回滚！")
        return False
    return True

myproject/
├── src/                  # 源代码
├── tests/                # 单元测试
│   ├── unit/
│   ├── integration/
│   └── ui/
├── scripts/              # 运维/部署脚本
├── docker/
│   └── Dockerfile
├── k8s/                  # 部署清单
├── docs/
├── pyproject.toml        # poetry 配置
├── .pre-commit-config.yaml
└── .gitlab-ci.yml / .github/workflows/

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_unstable_api():
    # ...