3.3. RESTful API 设计

良好的 API 设计是构建可维护、易用系统的关键。

3.3.1. REST 原则

3.3.1.1. 资源命名

# ✅ 好的 URL 设计
GET    /users              # 获取用户列表
GET    /users/123          # 获取特定用户
POST   /users              # 创建用户
PUT    /users/123          # 更新用户（完整替换）
PATCH  /users/123          # 部分更新用户
DELETE /users/123          # 删除用户

# 嵌套资源
GET    /users/123/orders           # 用户的订单
GET    /users/123/orders/456       # 特定订单
POST   /users/123/orders           # 创建订单

# ❌ 不好的设计
GET    /getUser/123                # 动词放 URL
GET    /users/123/getOrders        # 冗余
POST   /users/createUser           # 冗余

3.3.1.2. HTTP 方法语义

方法	用途	幂等性	安全性
GET	获取资源	是	是
POST	创建资源	否	否
PUT	完整更新	是	否
PATCH	部分更新	否	否
DELETE	删除资源	是	否

3.3.1.3. 状态码

from fastapi import HTTPException, status

# 成功响应
# 200 OK - 通用成功
# 201 Created - 创建成功
# 204 No Content - 删除成功（无返回内容）

@app.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    return new_user

@app.delete("/users/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(id: int):
    pass

# 客户端错误
# 400 Bad Request - 请求格式错误
# 401 Unauthorized - 未认证
# 403 Forbidden - 无权限
# 404 Not Found - 资源不存在
# 409 Conflict - 冲突（如重复创建）
# 422 Unprocessable Entity - 验证失败

# 服务端错误
# 500 Internal Server Error - 服务器错误
# 503 Service Unavailable - 服务不可用

3.3.2. 请求与响应设计

3.3.2.1. 统一响应格式

from pydantic import BaseModel
from typing import Generic, TypeVar, Optional, List

T = TypeVar('T')

class Response(BaseModel, Generic[T]):
    """统一响应格式"""
    success: bool
    data: Optional[T] = None
    error: Optional[str] = None
    
class PagedResponse(BaseModel, Generic[T]):
    """分页响应"""
    items: List[T]
    total: int
    page: int
    page_size: int
    pages: int

# 使用示例
@app.get("/users/{id}", response_model=Response[UserResponse])
async def get_user(id: int):
    user = await get_user_by_id(id)
    return Response(success=True, data=user)

@app.get("/users", response_model=PagedResponse[UserResponse])
async def list_users(page: int = 1, page_size: int = 20):
    users, total = await get_users_paginated(page, page_size)
    return PagedResponse(
        items=users,
        total=total,
        page=page,
        page_size=page_size,
        pages=(total + page_size - 1) // page_size
    )

3.3.2.2. 错误响应

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List, Optional

class ErrorDetail(BaseModel):
    field: Optional[str] = None
    message: str

class ErrorResponse(BaseModel):
    error: str
    code: str
    details: Optional[List[ErrorDetail]] = None

# 自定义异常
class APIError(Exception):
    def __init__(self, message: str, code: str, status_code: int = 400):
        self.message = message
        self.code = code
        self.status_code = status_code

# 异常处理器
@app.exception_handler(APIError)
async def api_error_handler(request: Request, exc: APIError):
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            error=exc.message,
            code=exc.code
        ).dict()
    )

# Pydantic 验证错误处理
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
    details = [
        ErrorDetail(field=".".join(map(str, err["loc"])), message=err["msg"])
        for err in exc.errors()
    ]
    return JSONResponse(
        status_code=422,
        content=ErrorResponse(
            error="Validation failed",
            code="VALIDATION_ERROR",
            details=details
        ).dict()
    )

3.3.3. 版本控制

3.3.3.1. URL 版本

# 推荐：URL 路径版本
from fastapi import APIRouter

v1_router = APIRouter(prefix="/api/v1")
v2_router = APIRouter(prefix="/api/v2")

@v1_router.get("/users")
async def get_users_v1():
    return {"version": "v1"}

@v2_router.get("/users")
async def get_users_v2():
    return {"version": "v2", "new_field": "value"}

app.include_router(v1_router)
app.include_router(v2_router)

3.3.3.2. 请求头版本

from fastapi import Header, Depends

async def get_api_version(
    api_version: str = Header("v1", alias="X-API-Version")
):
    return api_version

@app.get("/users")
async def get_users(version: str = Depends(get_api_version)):
    if version == "v2":
        return {"new_format": True}
    return {"old_format": True}

3.3.4. 过滤、排序、分页

from fastapi import Query
from typing import List, Optional
from enum import Enum

class SortOrder(str, Enum):
    asc = "asc"
    desc = "desc"

@app.get("/users")
async def list_users(
    # 分页
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    
    # 过滤
    status: Optional[str] = Query(None),
    created_after: Optional[datetime] = Query(None),
    search: Optional[str] = Query(None),
    
    # 排序
    sort_by: str = Query("created_at"),
    sort_order: SortOrder = Query(SortOrder.desc),
):
    query = User.query
    
    # 应用过滤
    if status:
        query = query.filter(User.status == status)
    if created_after:
        query = query.filter(User.created_at >= created_after)
    if search:
        query = query.filter(
            User.name.ilike(f"%{search}%") |
            User.email.ilike(f"%{search}%")
        )
    
    # 应用排序
    order_col = getattr(User, sort_by)
    if sort_order == SortOrder.desc:
        order_col = order_col.desc()
    query = query.order_by(order_col)
    
    # 分页
    total = query.count()
    users = query.offset((page - 1) * page_size).limit(page_size).all()
    
    return {
        "items": users,
        "total": total,
        "page": page,
        "page_size": page_size
    }

3.3.5. HATEOAS（超媒体）

from pydantic import BaseModel
from typing import Dict, List

class Link(BaseModel):
    href: str
    rel: str
    method: str = "GET"

class UserWithLinks(BaseModel):
    id: int
    name: str
    email: str
    _links: Dict[str, Link]

@app.get("/users/{id}")
async def get_user(id: int, request: Request):
    user = await get_user_by_id(id)
    
    base_url = str(request.base_url)
    
    return UserWithLinks(
        **user.dict(),
        _links={
            "self": Link(href=f"{base_url}users/{id}", rel="self"),
            "orders": Link(href=f"{base_url}users/{id}/orders", rel="orders"),
            "update": Link(href=f"{base_url}users/{id}", rel="update", method="PUT"),
            "delete": Link(href=f"{base_url}users/{id}", rel="delete", method="DELETE"),
        }
    )

3.3.6. API 文档

3.3.6.1. OpenAPI 增强

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="""
    ## 功能特性
    
    * **用户管理** - 创建、查询、更新、删除用户
    * **订单处理** - 订单相关操作
    
    ## 认证
    
    使用 Bearer Token 认证，在请求头中添加：
    ```
    Authorization: Bearer <token>
    ```
    """,
    version="1.0.0",
    contact={
        "name": "API Support",
        "email": "support@example.com"
    },
    license_info={
        "name": "MIT"
    }
)

# 为端点添加详细文档
@app.get(
    "/users/{user_id}",
    summary="获取用户详情",
    description="根据用户 ID 获取用户的详细信息",
    response_description="用户详细信息",
    responses={
        200: {"description": "成功获取用户"},
        404: {"description": "用户不存在"},
    }
)
async def get_user(user_id: int):
    """
    获取指定用户的详细信息：
    
    - **user_id**: 用户的唯一标识符
    
    返回用户的完整信息，包括：
    - 基本信息（姓名、邮箱等）
    - 创建时间
    - 最后更新时间
    """
    pass

3.3.7. 最佳实践

设计原则

一致性：URL、响应格式、错误处理保持一致
向后兼容：使用版本控制，避免破坏性变更
自描述：良好的命名和文档
无状态：每个请求包含所需的全部信息

安全考虑

始终使用 HTTPS
验证所有输入
限制响应数据（不返回敏感字段）
实施速率限制
使用适当的认证机制

性能优化

支持条件请求（ETag, Last-Modified）
实现分页
支持字段选择（稀疏字段集）
使用缓存
压缩响应（gzip）