Python项目稳定的最佳实践有哪些

本文目录导读：

1. 依赖管理
2. 代码质量
3. 测试策略
4. 错误处理
5. 日志记录
6. 配置管理
7. CI/CD 最佳实践
8. 性能与可靠性
9. 监控与告警
10. 部署安全

Python项目稳定性涉及代码质量、依赖管理、测试、部署和监控等多个方面,以下是一些经过验证的最佳实践：

依赖管理

锁定依赖版本

# requirements.txt
flask==2.3.0
requests==2.31.0
# 或使用 Pipfile/Pipfile.lock（Pipenv）
# 或 poetry.lock（Poetry）

使用虚拟环境

python -m venv .venv
source .venv/bin/activate  # Linux/Mac
.venv\Scripts\activate     # Windows

代码质量

类型提示

from typing import Optional
def process_user(user_id: int, name: Optional[str] = None) -> dict:
    """处理用户数据"""
    return {"id": user_id, "name": name or "unknown"}

代码格式化与检查

# 安装工具
pip install black flake8 mypy
# 自动格式化
black .
# 代码检查
flake8 src/
mypy src/

测试策略

单元测试

# tests/test_user_service.py
import pytest
from src.services import UserService
def test_create_user():
    service = UserService()
    result = service.create_user("test@example.com")
    assert result.id is not None

覆盖率要求

# 安装
pip install pytest-cov
# 运行并检查覆盖率
pytest --cov=src/ --cov-fail-under=80

集成测试

# tests/integration/test_api.py
def test_api_flow(client):
    # 创建用户
    response = client.post("/api/users", json={"email": "test@test.com"})
    assert response.status_code == 201
    # 查询用户
    user_id = response.json["id"]
    response = client.get(f"/api/users/{user_id}")
    assert response.status_code == 200

错误处理

使用自定义异常

class AppError(Exception):
    """应用基础异常"""
    pass
class NotFoundError(AppError):
    """资源不存在"""
    pass
def get_user(user_id: int) -> dict:
    if user_id <= 0:
        raise ValueError("Invalid user ID")
    # ... 业务逻辑

统一错误处理

# Flask 示例
@app.errorhandler(404)
def not_found(error):
    return {"error": "Resource not found"}, 404
@app.errorhandler(AppError)
def app_error(error):
    return {"error": str(error)}, 500

日志记录

结构化日志

import logging
import json
class StructuredLogger:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    def info(self, message: str, **kwargs):
        self.logger.info(json.dumps({
            "message": message,
            "data": kwargs
        }))
logger = StructuredLogger()
logger.info("User created", user_id=123, action="create")

配置管理

环境变量 + 配置文件

# config.py
import os
from dataclasses import dataclass
@dataclass
class Config:
    DATABASE_URL: str = os.getenv("DATABASE_URL", "sqlite:///dev.db")
    DEBUG: bool = os.getenv("DEBUG", "false").lower() == "true"
    LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")
config = Config()

CI/CD 最佳实践

GitHub Actions 示例

name: CI
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      - run: pip install -r requirements.txt
      - run: pip install pytest flake8 mypy
      - run: flake8 src/
      - run: mypy src/
      - run: pytest --cov=src/

性能与可靠性

重试机制

import time
from functools import wraps
def retry(max_attempts=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(delay)
            return wrapper
    return decorator
@retry(max_attempts=3, delay=2)
def unstable_api_call():
    # 可能失败的外部 API 调用
    pass

限流

from functools import wraps
import time
def rate_limit(calls_per_second=10):
    min_interval = 1.0 / calls_per_second
    def decorator(func):
        last_called = [0.0]
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left = min_interval - elapsed
            if left > 0:
                time.sleep(left)
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
        return wrapper
    return decorator

监控与告警

健康检查端点

@app.route("/health")
def health_check():
    return {
        "status": "ok",
        "timestamp": time.time(),
        "version": __version__
    }

关键指标收集

# 使用 prometheus_client 或 structlog
from prometheus_client import Counter, Histogram
REQUEST_COUNT = Counter('app_requests_total', 'Total requests')
REQUEST_DURATION = Histogram('app_request_duration_seconds', 'Request duration')
@app.route("/api/data")
def get_data():
    REQUEST_COUNT.inc()
    with REQUEST_DURATION.time():
        # 业务逻辑
        pass

部署安全

安全配置

# 不使用硬编码密码
import os
SECRET_KEY = os.environ.get("SECRET_KEY")
if not SECRET_KEY:
    raise RuntimeError("SECRET_KEY environment variable required")

依赖漏洞扫描

# 使用安全工具
pip install safety bandit
# 检查已知漏洞
safety check -r requirements.txt
# 静态安全分析
bandit -r src/

完整项目结构示例

my_project/
├── src/
│   ├── __init__.py
│   ├── config.py
│   ├── utils/
│   ├── services/
│   └── api/
├── tests/
│   ├── unit/
│   ├── integration/
│   └── fixtures/
├── scripts/
│   ├── deploy.sh
│   └── migrate.py
├── requirements.txt
├── setup.py
├── Dockerfile
├── .env.example
├── .flake8
├── pyproject.toml
└── README.md

稳定性是一个持续改进的过程，建议从基础的版本控制和测试开始,逐步引入更复杂的最佳实践。