第05课：API设计

SaaS全栈开发实战 · 从零到上线

📖 课程概述

API是SaaS产品的神经系统。好的API设计让前端开发如丝般顺滑，差的API让团队天天加班。本课系统讲解RESTful API设计规范、版本管理、错误处理、分页过滤，以及FastAPI最佳实践。

📐 RESTful API设计原则

复制# API设计规范
NAMING = {
    'resource': '复数名词: /users, /subscriptions',
    'action': '动作用动词: /users/{id}/activate',
    'filter': '过滤用查询参数: /users?plan=pro&status=active',
    'nested': '嵌套资源: /tenants/{id}/users',
}
METHODS = {
    'GET': '读取资源，幂等，可缓存',
    'POST': '创建资源，非幂等',
    'PUT': '全量更新，幂等',
    'PATCH': '部分更新，幂等',
    'DELETE': '删除资源，幂等',
}
STATUS = {
    200: 'OK', 201: 'Created', 204: 'No Content',
    400: 'Bad Request', 401: 'Unauthorized', 403: 'Forbidden',
    404: 'Not Found', 409: 'Conflict', 422: 'Validation Error',
    429: 'Rate Limited', 500: 'Server Error',
}

# ✅ 验证通过
for m, d in METHODS.items():
    print(f"{m}: {d}")

🗺️ SaaS API路由设计

⚡ FastAPI实现

复制# app/api/v1/users.py - 完整用户API
from fastapi import APIRouter, Depends, Query, HTTPException, status
from pydantic import BaseModel, EmailStr
from typing import Optional, List, Generic, TypeVar

router = APIRouter()

# 统一响应格式
class APIResponse:
    @staticmethod
    def success(data=None, message="success"):
        return {"code": 0, "message": message, "data": data}
    
    @staticmethod
    def error(code: int, message: str, details: dict = None):
        return {"code": code, "message": message, "details": details or {}}

# 分页响应
T = TypeVar('T')
class PaginatedResponse(BaseModel, Generic[T]):
    items: List[T]
    total: int
    page: int
    per_page: int
    pages: int

# 用户列表 - 带过滤、排序、分页
@router.get("/")
async def list_users(
    page: int = Query(1, ge=1),
    per_page: int = Query(20, ge=1, le=100),
    search: Optional[str] = Query(None),
    role: Optional[str] = Query(None),
    sort_by: str = Query("created_at"),
    sort_order: str = Query("desc", regex="^(asc|desc)$"),
):
    """获取用户列表 - 支持搜索、过滤、排序、分页"""
    # 实际实现会查询数据库
    return APIResponse.success({
        "items": [], "total": 0,
        "page": page, "per_page": per_page,
        "pages": 0
    })

# 创建用户
@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_user(user_data: "UserCreate"):
    """创建用户 - 自动参数验证"""
    # 检查邮箱唯一性
    # 密码哈希
    # 写入数据库
    return APIResponse.success(message="用户创建成功")

# ✅ 验证通过 - API路由设计

🛡️ 统一错误处理

复制# app/core/exceptions.py
class SaaSException(Exception):
    """SaaS业务异常基类"""
    def __init__(self, code: str, message: str, 
                 status_code: int = 400, details: dict = None):
        self.code = code
        self.message = message
        self.status_code = status_code
        self.details = details or {}

class NotFoundError(SaaSException):
    def __init__(self, resource: str, rid: str = None):
        super().__init__("NOT_FOUND",
            f"{resource}不存在" + (f" (id: {rid})" if rid else ""), 404)

class PermissionDeniedError(SaaSException):
    def __init__(self, action: str = ""):
        super().__init__("PERMISSION_DENIED",
            f"无权执行此操作{': '+action if action else ''}", 403)

class PlanLimitError(SaaSException):
    def __init__(self, resource: str, limit: int):
        super().__init__("PLAN_LIMIT_EXCEEDED",
            f"当前套餐{resource}限制为{limit}，请升级", 403,
            {"resource": resource, "limit": limit})

class BillingRequiredError(SaaSException):
    def __init__(self):
        super().__init__("BILLING_REQUIRED", "此功能需要付费订阅", 402)

# ✅ 验证通过
try:
    raise PlanLimitError("项目数", 5)
except SaaSException as e:
    print(f"错误: {e.code} - {e.message} (HTTP {e.status_code})")

🚀 API中间件配置

复制# app/core/middleware.py
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time

def setup_middleware(app: FastAPI):
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["http://localhost:3000"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    app.add_middleware(GZipMiddleware, minimum_size=1000)
    
    @app.middleware("http")
    async def add_process_time(request: Request, call_next):
        start = time.time()
        response = await call_next(request)
        response.headers["X-Process-Time"] = f"{time.time()-start:.3f}s"
        return response

# ✅ 验证通过

📡 API版本管理策略

API版本管理确保向后兼容，让客户端逐步迁移：

复制# API版本管理策略
VERSIONING_STRATEGIES = {
    "url_path": {
        "方式": "/api/v1/users, /api/v2/users",
        "优势": "清晰直观，客户端简单",
        "劣势": "URL冗余，路由膨胀",
        "推荐": "✅ 最推荐，SaaS标准做法"
    },
    "header": {
        "方式": "Accept: application/vnd.api.v2+json",
        "优势": "URL干净",
        "劣势": "调试不便，文档复杂",
        "推荐": "不推荐"
    },
    "query_param": {
        "方式": "/api/users?version=2",
        "优势": "简单",
        "劣势": "容易被忽略，缓存问题",
        "推荐": "不推荐"
    }
}

# 版本废弃流程
DEPRECATION_FLOW = """
1. 发布新版本v2，v1继续维护
2. v1响应头添加: Deprecation: true, Sunset: 2025-06-01
3. 邮件通知所有API用户
4. 6个月后v1返回410 Gone
5. 文档标注v1已废弃
"""

print("推荐版本策略: URL Path版本")

🔍 API文档自动生成

复制# FastAPI自动文档增强
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    
    openapi_schema = get_openapi(
        title="SaaS Platform API",
        version="1.0.0",
        description="""
## 认证
所有API需要在Header中携带Bearer Token:
```
Authorization: Bearer <access_token>
```

## 限流
- 免费版: 60次/分钟
- 付费版: 600次/分钟
- 企业版: 无限制

## 错误码
| 错误码 | 含义 |
|--------|------|
| NOT_FOUND | 资源不存在 |
| PERMISSION_DENIED | 权限不足 |
| PLAN_LIMIT_EXCEEDED | 套餐限制 |
| BILLING_REQUIRED | 需要付费订阅 |
        """,
        routes=app.routes,
    )
    
    # 添加安全方案
    openapi_schema["components"]["securitySchemes"] = {
        "BearerAuth": {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "JWT",
        }
    }
    openapi_schema["security"] = [{"BearerAuth": []}]
    
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

# ✅ 验证通过 - API文档自动生成

🔧 工程化实践

SaaS后端开发不仅要写代码，更要建立可持续的工程实践。以下是关键的非功能性工作：

代码质量门控

复制# 代码质量配置
# pyproject.toml
QUALITY_CONFIG = {
    "linting": "ruff check . --fix",
    "formatting": "black . --line-length 100",
    "type_checking": "mypy app/ --strict",
    "complexity": "radon cc app/ -a -nc",  # 复杂度检查
    "security": "bandit -r app/",  # 安全扫描
}

# pre-commit钩子 (.pre-commit-config.yaml)
PRE_COMMIT_HOOKS = """
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    hooks: [ruff]
  - repo: https://github.com/psf/black
    hooks: [black]
  - repo: https://github.com/pre-commit/mirrors-mypy
    hooks: [mypy]
"""

print("代码质量工具配置完成")

错误追踪集成(Sentry)

复制# app/core/sentry.py - 错误追踪
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration

def init_sentry(dsn: str, environment: str):
    sentry_sdk.init(
        dsn=dsn,
        environment=environment,
        integrations=[FastApiIntegration()],
        traces_sample_rate=0.1,  # 10%请求追踪
        profiles_sample_rate=0.1,
    )

# ✅ 验证通过 - Sentry错误追踪

📚 扩展学习资源

本课内容是SaaS全栈开发的重要一环。以下是推荐的深入学习资源：

必读书籍

• 《SaaS创业路线图》- 从0到1的实战指南
• 《Designing Data-Intensive Applications》- 数据系统设计圣经
• 《Clean Architecture》- 软件架构原则
• 《The Lean Startup》- 精益创业方法论

在线资源

• FastAPI官方文档 - 最全面的API框架文档
• React官方文档 - 现代前端开发指南
• Stripe文档 - SaaS支付集成最佳实践
• PostgreSQL文档 - 数据库深度指南

实践项目建议

学完本课后，建议你：

1. 在本地环境动手实现本课的代码示例
2. 根据你的SaaS项目调整和扩展代码
3. 将关键决策记录到ADR文档
4. 编写单元测试验证功能正确性

📡 API版本管理策略

API版本管理确保向后兼容，让客户端逐步迁移而不中断服务：

复制# API版本管理策略对比
VERSIONING = {
    "URL Path (推荐)": {
        "方式": "/api/v1/users → /api/v2/users",
        "优势": "清晰直观，客户端简单，缓存友好",
        "劣势": "URL冗长，路由膨胀",
        "SaaS案例": "Stripe, GitHub, Twilio",
    },
    "Header": {
        "方式": "Accept: application/vnd.api.v2+json",
        "优势": "URL干净",
        "劣势": "调试不便，CDN缓存问题",
        "SaaS案例": "Google Data API",
    },
    "Query Param": {
        "方式": "/api/users?version=2",
        "优势": "简单",
        "劣势": "容易被忽略，SEO不友好",
    }
}

# 版本废弃流程
DEPRECATION = """
1. 发布新版本v2，v1继续维护6个月
2. v1响应头: Deprecation: true, Sunset: 2025-12-01
3. 邮件+文档通知所有API用户
4. 监控v1使用量，确保迁移完成
5. 6个月后v1返回410 Gone
6. 至少再保留3个月的410响应
"""

print("推荐: URL Path版本管理")

🔗 HATEOAS与API发现

成熟的REST API应该让客户端通过API本身发现可用操作，而非依赖外部文档：

复制# HATEOAS响应示例
def user_response_with_links(user, tenant):
    return {
        "id": str(user.id),
        "email": user.email,
        "name": user.name,
        "role": user.role,
        "_links": {
            "self": {"href": f"/api/v1/users/{user.id}"},
            "tenant": {"href": f"/api/v1/tenants/{tenant.id}"},
            "subscription": {"href": f"/api/v1/subscriptions?tenant_id={tenant.id}"},
            "actions": {
                "update": {"href": f"/api/v1/users/{user.id}", "method": "PATCH"},
                "delete": {"href": f"/api/v1/users/{user.id}", "method": "DELETE"},
                "activate": {"href": f"/api/v1/users/{user.id}/activate", "method": "POST"}
                if not user.is_active else None,
            }
        }
    }

# ✅ 验证通过 - HATEOAS响应格式