🐍 Python 第五课：传说之境——工程化、架构与部署

课程目标：

• 掌握测试驱动开发（TDD），让代码质量有保障
• 学会类型提示，让代码自文档化
• 代码规范与自动化工具，告别"屎山"
• Docker 容器化部署，一次构建到处运行
• 性能优化与Profiling，找到瓶颈
• 设计模式与架构思想，写出"优雅"的代码
• 微服务入门，为大型系统做准备

1) 回顾上节课

上节课我们学会了：

• 生成器与迭代器：省内存的神器
• 装饰器：给函数"开挂"
• 上下文管理器：资源管理更优雅
• 并发编程：多线程、多进程、异步IO
• 数据库操作：SQLite与SQLAlchemy
• Web开发：Flask框架

常见问题解答：

Q：异步代码里能直接用time.sleep()吗？
A：不行！time.sleep()会阻塞整个事件循环。要用await asyncio.sleep()，这样等待时可以把控制权交给其他任务。

Q：SQLAlchemy的session什么时候关闭？
A：推荐用上下文管理器：with Session() as session:，或者确保在请求结束时关闭，避免连接泄漏。

2) 测试驱动开发（TDD）：先写测试，再写代码

为什么要测试？

• 手动测试的痛苦：改一行代码，手动测10个功能，漏了一个就线上故障
• 重构的勇气：有测试覆盖，才敢大胆重构
• 文档作用：测试用例就是最佳使用示例

pytest：Python测试之王

# 安装
pip install pytest pytest-cov

基本用法：

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

def divide(a, b):
    if b == 0:
        raise ValueError("不能除以零")
    return a / b

# test_calculator.py
import pytest
from calculator import add, divide

def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0
    assert add(0.1, 0.2) == pytest.approx(0.3)  # 浮点数比较

def test_divide():
    assert divide(10, 2) == 5
    assert divide(7, 2) == 3.5

def test_divide_by_zero():
    with pytest.raises(ValueError, match="不能除以零"):
        divide(10, 0)

运行测试：

pytest -v                    # 详细输出
pytest -s                    # 显示print输出
pytest --cov=calculator      # 覆盖率报告
pytest -k "add"              # 只运行名字包含add的测试

Fixture：测试的"脚手架"

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base, User

@pytest.fixture
def db_session():
    """每个测试用例都使用全新的内存数据库"""
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    
    yield session  # 提供session给测试用例
    
    session.close()  # 测试结束后清理

def test_create_user(db_session):
    user = User(name="张三", email="zhangsan@example.com")
    db_session.add(user)
    db_session.commit()
    
    result = db_session.query(User).first()
    assert result.name == "张三"

Mock：模拟外部依赖

from unittest.mock import Mock, patch
import requests

def get_weather(city):
    """获取天气（实际会调用外部API）"""
    resp = requests.get(f"https://api.weather.com/{city}")
    return resp.json()

def test_get_weather():
    # 模拟requests.get，避免真实网络请求
    mock_response = Mock()
    mock_response.json.return_value = {"temp": 25, "city": "北京"}
    
    with patch('requests.get', return_value=mock_response):
        result = get_weather("北京")
        assert result["temp"] == 25

TDD实战：Red-Green-Refactor

流程：

1. Red：先写测试（会失败，因为功能还没实现）
2. Green：写最少代码让测试通过
3. Refactor：重构代码，保持测试通过

# 第一步：写测试（Red）
def test_password_strength():
    assert is_strong_password("123") == False
    assert is_strong_password("abcdef") == False
    assert is_strong_password("Abcdef1!") == True  # 大写+小写+数字+特殊字符

# 第二步：实现功能（Green）
def is_strong_password(password):
    if len(password) < 8:
        return False
    has_upper = any(c.isupper() for c in password)
    has_lower = any(c.islower() for c in password)
    has_digit = any(c.isdigit() for c in password)
    has_special = any(c in "!@#$%^&*" for c in password)
    return all([has_upper, has_lower, has_digit, has_special])

# 第三步：重构（保持测试通过）
# 可以提取验证规则，让代码更可扩展

3) 类型提示：让代码自文档化

基础类型

from typing import List, Dict, Optional, Union, Tuple

def greet(name: str) -> str:
    return f"Hello, {name}"

def process_data(items: List[int]) -> Dict[str, int]:
    return {"count": len(items), "sum": sum(items)}

def find_user(user_id: int) -> Optional[User]:
    """可能返回None"""
    return session.query(User).get(user_id)

def parse_value(value: Union[str, int]) -> float:
    """接受字符串或整数，返回浮点数"""
    return float(value)

复杂类型

from typing import Callable, Iterator, TypeVar, Generic

# 类型变量
T = TypeVar('T')

def first(items: List[T]) -> Optional[T]:
    return items[0] if items else None

# 可调用对象
def executor(func: Callable[[int, int], int], a: int, b: int) -> int:
    return func(a, b)

# 生成器类型
def fibonacci(n: int) -> Iterator[int]:
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b

类与泛型

from typing import Generic, TypeVar

T = TypeVar('T')

class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: List[T] = []
    
    def push(self, item: T) -> None:
        self._items.append(item)
    
    def pop(self) -> T:
        if not self._items:
            raise IndexError("Stack is empty")
        return self._items.pop()

# 使用
int_stack: Stack[int] = Stack()
int_stack.push(42)
# int_stack.push("hello")  # 类型检查器会报错！

类型检查工具

# mypy：静态类型检查
pip install mypy
mypy my_project/  # 检查整个项目

# 集成到编辑器（VS Code）
# 安装Python插件，开启类型检查

4) 代码质量：从"能跑"到"优雅"

代码格式化：Black

pip install black
black my_project/  # 一键格式化，无需争论风格

Black的原则：

• 单引号变双引号
• 行长度限制88字符（可配置）
• 统一的缩进和换行
• "不妥协的格式化"，减少团队争论

代码检查：Flake8 + isort

pip install flake8 isort

flake8 my_project/  # 检查PEP8规范、语法错误
isort my_project/   # 自动排序import

配置.flake8：

[flake8]
max-line-length = 88
extend-ignore = E203, W503  # 与Black兼容
exclude = .git,__pycache__,venv

Pre-commit：提交前自动检查

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/PyCQA/isort
    rev: 5.12.0
    hooks:
      - id: isort

安装钩子：

pre-commit install  # 以后每次git commit都会自动检查

5) Docker容器化：一次构建，到处运行

Dockerfile编写

# 使用官方Python基础镜像
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

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

# 复制应用代码
COPY . .

# 暴露端口
EXPOSE 5000

# 运行应用
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "app:app"]

.dockerignore：

__pycache__
*.pyc
.env
.git
venv/
.pytest_cache/

Docker Compose：多容器编排

# docker-compose.yml
version: '3.8'

services:
  web:
    build: .
    ports:
      - "5000:5000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
    depends_on:
      - db
      - redis
  
  db:
    image: postgres:15
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  redis:
    image: redis:7-alpine
  
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - web

volumes:
  postgres_data:

常用命令：

docker-compose up -d          # 后台启动所有服务
docker-compose logs -f web    # 查看web服务日志
docker-compose exec web bash  # 进入web容器
docker-compose down           # 停止并删除容器

6) 性能优化：找到瓶颈，精准打击

Profiling：性能分析

# 使用cProfile
import cProfile
import pstats

def slow_function():
    total = 0
    for i in range(1000000):
        total += i ** 2
    return total

# 方式1：命令行
# python -m cProfile -s cumulative my_script.py

# 方式2：代码内
profiler = cProfile.Profile()
profiler.enable()
result = slow_function()
profiler.disable()

stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10)  # 打印前10个最耗时的函数

line_profiler：逐行分析

pip install line_profiler

from line_profiler import profile

@profile
def heavy_computation():
    result = []
    for i in range(10000):
        temp = [x ** 2 for x in range(1000)]  # 这行慢？
        result.append(sum(temp))
    return result

# 运行：kernprof -l -v script.py

优化策略

缓存示例：

from functools import lru_cache
import time

@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# 第一次计算fib(30)慢，之后瞬间返回
start = time.time()
print(fibonacci(30))  # 832040
print(f"耗时：{time.time() - start:.4f}秒")

start = time.time()
print(fibonacci(30))  # 瞬间完成，从缓存读取
print(f"耗时：{time.time() - start:.4f}秒")

7) 设计模式：前辈的智慧

单例模式：只有一个实例

# 方式1：模块天然单例
# singleton.py
class Database:
    def __init__(self):
        self.connection = "连接池"

db = Database()  # 导入时创建

# 其他文件：from singleton import db，永远是同一个对象

# 方式2：装饰器
def singleton(cls):
    instances = {}
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance

@singleton
class Logger:
    def __init__(self):
        self.level = "INFO"

工厂模式：创建对象的艺术

from abc import ABC, abstractmethod

class Animal(ABC):
    @abstractmethod
    def speak(self):
        pass

class Dog(Animal):
    def speak(self):
        return "汪汪"

class Cat(Animal):
    def speak(self):
        return "喵喵"

class AnimalFactory:
    @staticmethod
    def create(animal_type: str) -> Animal:
        animals = {
            "dog": Dog,
            "cat": Cat
        }
        if animal_type not in animals:
            raise ValueError(f"未知动物：{animal_type}")
        return animals[animal_type]()

# 使用
dog = AnimalFactory.create("dog")
print(dog.speak())  # 汪汪

观察者模式：事件订阅

from typing import List, Callable

class EventManager:
    def __init__(self):
        self._listeners: Dict[str, List[Callable]] = {}
    
    def subscribe(self, event_type: str, listener: Callable):
        if event_type not in self._listeners:
            self._listeners[event_type] = []
        self._listeners[event_type].append(listener)
    
    def notify(self, event_type: str, data):
        for listener in self._listeners.get(event_type, []):
            listener(data)

# 使用
def send_email(user):
    print(f"发送邮件给 {user['email']}")

def send_sms(user):
    print(f"发送短信给 {user['phone']}")

event_manager = EventManager()
event_manager.subscribe("user_registered", send_email)
event_manager.subscribe("user_registered", send_sms)

# 用户注册后
event_manager.notify("user_registered", {
    "email": "user@example.com",
    "phone": "13800138000"
})

8) 微服务架构入门

单体 vs 微服务

服务拆分示例：电商系统

monolithic/                 microservices/
├── app.py                  ├── user-service/     (用户服务)
├── models.py               │   ├── app.py
├── views.py                │   └── Dockerfile
└── templates/              ├── order-service/    (订单服务)
                            │   ├── app.py
                            ├── payment-service/  (支付服务)
                            │   └── ...
                            └── api-gateway/      (API网关)
                                └── nginx.conf

服务间通信

REST API：

# user-service 提供用户查询
@app.route('/api/users/<int:id>')
def get_user(id):
    return jsonify(User.query.get_or_404(id).to_dict())

# order-service 调用 user-service
import requests

def create_order(user_id, items):
    # 查询用户信息
    resp = requests.get(f"http://user-service:5000/api/users/{user_id}")
    user = resp.json()
    
    # 创建订单...
    return Order(user_id=user_id, items=items)

消息队列（RabbitMQ/Celery）：异步解耦

# 订单服务发布事件
from celery import Celery

celery = Celery('tasks', broker='redis://redis:6379/0')

@celery.task
def send_order_notification(order_id):
    # 异步发送通知，不阻塞主流程
    order = Order.query.get(order_id)
    send_email(order.user.email, "订单已创建")

# 创建订单后调用
send_order_notification.delay(order.id)

9) 实战项目：完整的企业级应用

项目：微服务架构的任务调度系统

task-scheduler/
├── docker-compose.yml
├── nginx/
├── services/
│   ├── api-gateway/          # API网关（Nginx/FastAPI）
│   ├── auth-service/         # 认证服务（JWT）
│   ├── task-service/         # 任务管理服务
│   ├── worker-service/       # 任务执行器（Celery）
│   └── notification-service/ # 通知服务
├── shared/
│   └── models/               # 共享模型定义
└── tests/
    └── integration/          # 集成测试

核心代码示例：

# task-service/app.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
import schemas, models
from database import get_db
from celery import Celery

app = FastAPI(title="Task Service")
celery_app = Celery('tasks', broker='redis://redis:6379/0')

@app.post("/tasks/", response_model=schemas.Task)
def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db)):
    db_task = models.Task(**task.dict())
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    
    # 异步执行任务
    execute_task.delay(db_task.id)
    
    return db_task

@app.get("/tasks/", response_model=List[schemas.Task])
def list_tasks(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return db.query(models.Task).offset(skip).limit(limit).all()

@celery_app.task
def execute_task(task_id: int):
    """Celery任务：实际执行工作"""
    # 更新状态为运行中
    update_task_status(task_id, "running")
    
    try:
        # 执行实际工作...
        time.sleep(10)  # 模拟耗时操作
        update_task_status(task_id, "completed")
        notify_completion.delay(task_id)
    except Exception as e:
        update_task_status(task_id, "failed", str(e))
        raise

10) 学习路线与资源

进阶书单

优秀开源项目学习

• Flask：轻量级Web框架，源码易读
• Celery：分布式任务队列
• FastAPI：现代异步Web框架
• Django：全功能Web框架
• Scrapy：爬虫框架，学习架构设计

持续精进

1. 写博客：把学到的教给别人，是最好的学习
2. 参与开源：从提Issue、修文档开始
3. 技术分享：内部分享、技术会议
4. 关注PEP：了解Python最新发展

结语：从学徒到大师

Python学习之路：

入门 → 进阶 → 高级 → 专家 → 大师
 │      │      │      │      │
print  类与    装饰器  架构   创造
变量   对象    异步   设计   新范式
循环   数据库  测试   优化   
函数   Web    部署   领导力

记住：

• 代码是写给人看的，顺便给机器执行
• 简单优于复杂，明确优于隐晦
• 现在偷懒，将来还债；现在严谨，将来从容

最后送大家一句话：

"The Zen of Python: Simple is better than complex. Complex is better than complicated."

祝编程愉快！You are now a Python Legend! 🐍👑

毕业项目：
用所学知识，从零开始设计并实现一个完整的系统：

• 需求分析、架构设计
• TDD开发、类型提示
• Docker部署、CI/CD
• 性能优化、监控告警
• 技术文档、用户手册

这是你成为传说的证明！

课程结束，但学习永不停止！