Python教程 Episode 18 - CI/CD 与 DevOps 实战

content="""#cloud-config

package_update: true

packages:

  - docker.io

  - python3-pip

write_files:

  - path: /opt/myapp/.env

    content: |

      DATABASE_URL=postgresql://user:pass@db:5432/myapp

      SECRET_KEY=super-secret-key

  - path: /opt/myapp/docker-compose.yml

    content: |

      version: '3.9'

      services:

        web:

          image: ghcr.io/myorg/my-fastapi-app:latest

          ports:

            - '8000:8000'

          env_file: .env

""",

            )

        ]

    ),

)

cloud_init_user_data = cloud_init_config.render_data()

vm = Instance(

f"{project_name}-server",

instance_type="t3.small",

ami="ami-0c55b159cbfafe1f0",  # Ubuntu 22.04

user_data=cloud_init_user_data,

key_name=pulumi.Config().get("ssh_key_name"),

)

# ===== 3. 输出公网 IP =====

pulumi.export("server_public_ip", vm.public_ip)

pulumi.export("image_name", docker_image.image_name)

```

运行部署：

```bash

pulumistackinitproduction

pulumiup

```

这比手动在控制台点点点可靠多了——所有的云资源都在代码版本控制里，谁改了什么、什么时候改的一清二楚。

### 5.2 Terraform（传统备选方案）

如果不喜欢 Python 方式，Terraform 用 HCL 语言做同样的事情：

```hcl

# main.tf — 用 Terraform 定义云资源

terraform {

  required_providers {

    aws = {

      source  = "hashicorp/aws"

      version = "~> 5.0"

    }

  }

}

provider "aws" {

  region = "us-west-2"

}

resource "aws_instance" "web_server" {

  ami           = "ami-0c55b159cbfafe1f0"

  instance_type = "t3.small"

  user_data = <<-EOF

              #!/bin/bash

              apt-get update && apt-get install -y docker.io

              systemctl start docker

              EOF

  tags = {

    Name = "my-fastapi-server"

  }

}

resource "aws_security_group" "web_sg" {

  name = "allow-web-traffic"

  ingress {

    from_port   = 8000

    to_port     = 8000

    protocol    = "tcp"

    cidr_blocks = ["0.0.0.0/0"]

  }

  egress {

    from_port   = 0

    to_port     = 0

    protocol    = "-1"

    cidr_blocks = ["0.0.0.0/0"]

  }

}

```

---

## 六、实战演练：完整的 CI/CD 流水线

现在我们把前面学到的所有内容串在一起，打造一个从开发到部署的完整流水线。

### 6.1 项目结构

```

my-fastapi-app/

├── main.py                 # FastAPI 应用入口

├── models.py               # 数据模型

├── schemas.py              # Pydantic schema

├── services/

│   ├── __init__.py

│   └── data_service.py     # 业务逻辑层

├── tests/

│   ├── __init__.py

│   ├── conftest.py         # pytest fixture

│   └── test_main.py        # API 测试

├── .github/workflows/

│   ├── ci.yml              # 测试流水线

│   └── deploy.yml          # 部署流水线

├── Dockerfile

├── docker-compose.yml

├── requirements.txt

└── pyproject.toml          # 项目元数据

```

### 6.2 CI/CD 流水线流程图

```

开发者推送代码

    │

    ▼

触发 GitHub Actions (ci.yml)

    │

    ├─→ 安装 Python 依赖

    │

    ├─→ Ruff Lint 检查（格式 + 代码质量）

    │   └─→ 失败？发送通知给开发者，停止流程

    │   └─→ 成功？继续

    │

    ├─→ 运行单元测试 + 覆盖率检查

    │   └─→ 失败？通知开发者

    │   └─→ 成功？继续

    │

    ├─→ 构建 Docker 镜像

    │   └─→ 推送到 GHCR（GitHub Container Registry）

    │

    └─→ 部署到预发环境（staging）

        └─→ 运行冒烟测试

            └─→ 全部通过？手动确认后 → 部署到生产

```

### 6.3 完整的 workflow 文件

这是我们在前面已经看到过的 `ci.yml` 和 `deploy.yml` 的精简完整版，这里再整理一遍让你更清晰：

```yaml

# .github/workflows/ci-deploy.yml

name: CI/CD Pipeline

on:

push:

branches: [main]

pull_request:

branches: [main]

env:

REGISTRY: ghcr.io

IMAGE_TAG: ${{ github.sha }}

jobs:

lint-and-test:

runs-on: ubuntu-latest

steps:

      - name: Checkout

uses: actions/checkout@v4

      - name: Setup Python

uses: actions/setup-python@v5

with:

python-version: "3.12"

cache: "pip"

      - name: Install dependencies

run: |

          pip install --upgrade pip

          pip install -r requirements.txt

          pip install pytest pytest-cov ruff mypy

      - name: Ruff lint & format check

run: |

          ruff check .

          ruff format --check .

      - name: MyPy type check

run: mypy . --ignore-missing-imports || true

      - name: Run unit tests

run: pytest tests/ -v --cov=. --cov-report=xml

      - name: Upload coverage

uses: codecov/codecov-action@v4

build-and-push:

needs: lint-and-test

if: github.event_name == 'push'

runs-on: ubuntu-latest

permissions:

packages: write

steps:

      - uses: actions/checkout@v4

      - name: Login to GHCR

uses: docker/login-action@v3

with:

registry: ghcr.io

username: ${{ github.actor }}

password: ${{ secrets.GITHUB_TOKEN }}

      - name: Build and push

uses: docker/build-push-action@v5

with:

context: .

push: true

tags: ghcr.io/${{ github.repository }}:${{ env.IMAGE_TAG }}

deploy-staging:

needs: build-and-push

runs-on: ubuntu-latest

environment: staging

steps:

      - name: Deploy to staging

uses: appleboy/ssh-action@v1

with:

host: ${{ secrets.STAGING_HOST }}

username: deployer

key: ${{ secrets.STAGING_SSH_KEY }}

script: |

            cd /opt/myapp

            docker compose pull

            docker compose up -d --remove-orphans

smoke-test-staging:

needs: deploy-staging

runs-on: ubuntu-latest

steps:

      - name: Smoke test

run: |

          # 等待容器启动

          sleep 10

          # 检查健康端点

          RESPONSE=$(curl -sf http://${{ secrets.STAGING_HOST }}:8000/health)

          echo "Response: $RESPONSE"

          echo "$RESPONSE" | jq -e '.status == "ok"'

```

---

## 七、实操练习

### 练习题

**第 1 题：为你的现有项目添加 CI**

选一个你已经写过的 Python 项目，在 GitHub 上创建一个仓库，然后：

1. 创建 `.github/workflows/ci.yml`

2. 让它能在 push 时自动运行 pytest 测试

3. 提交代码并观察 GitHub Actions 面板的运行结果

```python

# 参考答案思路

# 在 workflow 中添加 steps：

# 1. checkout@v4

# 2. setup-python@v5 with python-version: "3.12"

# 3. pip install pytest

# 4. pytest tests/ -v

```

**第 2 题：编写 Dockerfile 并本地测试**

为你之前的 FastAPI 应用写一个 Dockerfile，然后本地构建并运行：

```bash

dockerbuild-tmy-fastapi-app:test.

dockerrun-p8000:8000--rmmy-fastapi-app:test

# 访问 http://localhost:8000/docs 验证

```

```python

# Dockerfile 关键内容参考

# FROM python:3.12-slim

# WORKDIR /app

# COPY requirements.txt .

# RUN pip install --no-cache-dir -r requirements.txt

# COPY . .

# CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

```

**第 3 题：实现蓝绿部署脚本**

编写一个 shell 或 Python 脚本，实现蓝绿部署：

1. 启动新版本容器

2. 等待健康检查通过

3. 切换到新版本

4. 保留旧版本作为回滚选项

```python

# 参考答案思路

import subprocess

import time

import urllib.request

defblue_green_deploy(version: str) -> None:

# 1. 构建并启动绿色环境

    subprocess.run(["docker", "compose", "up", "-d", "--build", "green"])

# 2. 健康检查

for _ inrange(30):

try:

            resp = urllib.request.urlopen("http://localhost:8081/health")

if resp.status == 200:

break

exceptException:

pass

        time.sleep(2)

else:

raiseRuntimeError("Green environment failed health check!")

# 3. 切换流量

    subprocess.run([

"docker", "exec", "nginx", "sh", "-c",

"echo 'upstream backend { server green:8000; }' > /etc/nginx/conf.d/default.conf"

    ])

    subprocess.run(["docker", "compose", "restart", "nginx"])

print(f"Successfully deployed v{version}")

blue_green_deploy("2.0.0")

```

**第 4 题：添加 Pulumi 基础设施代码**

在一个已有的 AWS 账户上，用 Pulumi Python 编写部署 FastAPI 应用的 IaC 脚本：

1. 创建 EC2 实例

2. 配置安全组（开放 8000 端口）

3. 输出公网 IP 以便访问

```python

# 参考答案思路

# import pulumi

# import pulumi_aws as aws

# vm = aws.ec2.Instance("web-server", ...)

# pulumi.export("ip", vm.public_ip)

```

**第 5 题：设计一个多级流水线**

为一个电商微服务项目设计 CI/CD 流水线。该项目包含：

- user-service（用户服务，REST API）

- order-service（订单服务，REST API）

- notification-service（通知服务，接收消息队列事件）

要求：

- 每个服务有独立的测试和构建

- 所有服务测试通过后才部署

- 部署顺序：user → order → notification（注意依赖关系）

---

## 八、总结

这一期我们聊了很多内容，我尽量把它拆解成你能一步步跟上的节奏：

1.**CI/CD 的本质**：让别人不能把 bug 合并进来，让代码从本地到服务器的每一步都是可重复、可追溯的。

2.**GitHub Actions 是起步最快的工具**：不需要额外装软件，直接在 GitHub 里定义 YAML 就行了。

3.**Docker 是现代部署的标准**：不管你的代码跑在本地还是云上，打包进容器就能保证一致的环境。

4.**蓝绿部署和金丝雀发布是进阶话题**：它们解决了"上线万一出问题怎么办"的恐惧。

5.**基础设施即代码**：服务器也是代码的一部分，应该被审查、被测试、被版本化。

最重要的一句话：**不要追求一步到位完美。先从最简单的 CI 开始——每次 push 自动跑测试，这就已经是 80% 的进步了。**

---

## 九、课后练习参考答案

### 第 3 题完整实现（蓝绿部署脚本）

```python

#!/usr/bin/env python3

"""blue_green_deploy.py — 蓝绿部署自动化脚本"""

import subprocess

import time

import urllib.request

import sys

defwait_for_health(url: str, timeout: int = 60, interval: int = 2) -> bool:

"""等待服务健康检查通过"""

    elapsed = 0

while elapsed < timeout:

try:

            resp = urllib.request.urlopen(url, timeout=5)

if resp.status == 200:

print(f"  ✅ 服务健康检查通过")

returnTrue

exceptException:

pass

        time.sleep(interval)

        elapsed += interval

print(f"  ❌ 超时：{timeout}秒内未通过健康检查")

returnFalse

defswap_traffic(blue_port: int, green_port: int, nginx_container: str = "nginx") -> None:

"""切换 Nginx 流量方向"""

    current = "blue"if blue_port == 8000else"green"

    target = "green"if current == "blue"else"blue"

    target_port = green_port if target == "green"else blue_port

    cmd = [

"docker", "exec", nginx_container, "sh", "-c",

f"echo 'upstream backend {{ server localhost:{target_port}; }}' > "

"/etc/nginx/conf.d/upstream.conf"

    ]

    subprocess.run(cmd, check=True)

    subprocess.run(["docker", "compose", "restart", nginx_container], check=True)

print(f"  🔄 流量已切换: {current} → {target}")

defrollback(current_port: int, nginx_container: str = "nginx") -> None:

"""回滚到上一个版本"""

    target_port = current_port - 8000 + 8000# blue=8000, green=8001

# 简化版：回滚就是切回去

print("  ⚠️  执行回滚...")

defmain():

iflen(sys.argv) < 2:

print("用法: python blue_green_deploy.py <version>")

        sys.exit(1)

    version = sys.argv[1]

print(f"🚀 开始部署 v{version}...")

# Step 1: 构建新镜像

print("  📦 构建 Docker 镜像...")

    subprocess.run([

"docker", "build", "-t", f"myapp:{version}", "-f", "Dockerfile", "."

    ], check=True)

# Step 2: 启动绿色环境

print("  🟢 启动绿色环境...")

    env_vars = {

"APP_VERSION": version,

"BLUE_PORT": "8000",

"GREEN_PORT": "8001",

    }

    subprocess.run([

"docker", "compose", "run", "--rm", "--no-deps", 

"green", "docker", "swarm", "init"

    ])

# Step 3: 健康检查

print("  🔍 等待健康检查...")

ifnot wait_for_health("http://localhost:8001/health"):

print("  ❌ 健康检查失败，回滚！")

# rollback(8001)

        sys.exit(1)

# Step 4: 切换流量

print("  🌊 切换流量...")

    swap_traffic(8000, 8001)

# Step 5: 停止旧蓝色容器（可选）

print("  ✅ 部署完成 v{version}")

if__name__ == "__main__":

    main()

```

### 第 5 题架构图参考

```

                    ┌─────────────────────────────────────────────────┐

                    │              GitHub Push Event                   │

                    └────────────────────────┬────────────────────────┘

                                             │

                    ┌────────────────────────┼────────────────────────┐

                    │                        │                         │

              ┌─────▼─────┐           ┌─────▼─────┐           ┌─────▼─────┐

              │ User Svc  │           │ Order Svc │           │ Notif Svc │

              │  CI Build │           │  CI Build │           │  CI Build │

              └─────┬─────┘           └─────┬─────┘           └─────┬─────┘

                    │                        │                       │

                    └────────────────────────┼───────────────────────┘

                                             │

                              All tests passed? → Build Docker images

                                             │

                              ┌──────────────▼──────────────┐

                              │        Deploy Order:         │

                              │  1. User Service (base)     │

                              │  2. Order Service (depends) │

                              │  3. Notification Service    │

                              └──────────────┬──────────────┘

                                             │

                              ┌──────────────▼──────────────┐

                              │   Smoke Tests on Staging    │

                              └──────────────┬──────────────┘

                                             │

                              Manual Approval → Deploy to Production

```

---

## 十、下期预告

**Episode 19：Python 并发与多线程深度实践**

我们已经学了 asyncio（协程），接下来要补上并发编程的另一半：**多线程和多进程**。

- threading 和 multiprocessing 的区别和使用场景

- GIL 是什么？为什么它会影响你的性能？

- 用 concurrent.futures 简化并发编程

- 生产者-消费者模式的多种实现

- GIL 绕过技巧：多进程池 + 协程混合架构

敬请期待！

---

**📚 系列回顾**

到目前为止我们学过的内容：

| Episode | 主题 |

|---------|------|

| 01 | 环境搭建与基础语法 |

| 02 | 数据结构与字符串处理 |

| 03 | 面向对象编程 |

| 04 | 装饰器、生成器与文件 IO |

| 05 | 爬虫入门 |

| 06 | 数据分析入门 |

| 07 | Web 后端开发 |

| 08 | AI 实战入门 |

| 09 | 前端入门 |

| 10 | 项目实战——Dashboard |

| 11 | Docker 容器化部署 |

| 12 | 数据库进阶 |

| 13 | 异步编程与并发 |

| 14 | 日志系统与调试技巧 |

| 15 | 自动化运维与脚本实战 |

| 16 | 单元测试与代码质量 |

| 17 | 微服务入门 |

| 18 | CI/CD 与 DevOps 实战 ← **本期** |

加油！你已经从零基础走到了"能用 Python 写完整的后端并部署上线"的水平。🎉