rest-api-design-patterns

REST API 设计模式

一套全面的技能，用于设计、实现和维护 RESTful API。掌握资源建模、HTTP 方法、版本控制策略、分页、筛选、错误处理，以及使用 FastAPI、Express.js 和现代框架构建可扩展、易维护 API 的最佳实践。

何时使用此技能

在以下情况下使用此技能：

• 从头开始设计新的 RESTful API
• 使用 HTTP/REST 接口构建微服务
• 重构现有 API，以改进设计并保持一致性
• 使用恰当的 HTTP 语义实现 CRUD 操作
• 为现有 API 添加版本控制
• 设计资源关系和嵌套端点
• 实现分页、筛选和排序
• 以一致的方式处理错误和验证
• 构建超媒体驱动的 API（HATEOAS）
• 使用缓存和压缩优化 API 性能
• 使用 OpenAPI/Swagger 规范编写 API 文档
• 使用身份验证和授权模式确保 API 安全

REST 核心原则

什么是 REST？

REST（表述性状态转移）是一种分布式系统架构风格，强调以下几点：

1. 基于资源：一切都是具有唯一标识符（URI）的资源
2. 标准方法：使用标准 HTTP 方法（GET、POST、PUT、DELETE、PATCH）
3. 无状态：每个请求都包含处理该请求所需的全部信息
4. 客户端-服务端：客户端与服务端明确分离
5. 可缓存：响应可以缓存以提升性能
6. 统一接口：整个 API 遵循一致的模式

REST 成熟度模型（Richardson 成熟度模型）

0 级 - POX 沼泽：单个 URI，单个 HTTP 方法（通常为 POST）

• 示例：通过 POST 请求体在 /api 中执行所有操作

1 级 - 资源：多个 URI，每个 URI 表示一个资源

• 示例：/users、/posts、/products

2 级 - HTTP 动词：正确使用 HTTP 方法

• 示例：GET /users/123、POST /users、PUT /users/123

3 级 - 超媒体控件（HATEOAS）：API 响应中包含相关资源的链接

• 示例：响应包含 "_links": {"self": "/users/123", "posts": "/users/123/posts"}

资源建模

资源命名约定

1. 使用名词，而非动词

Good:
  GET /users
  GET /products
  POST /orders

Bad:
  GET /getUsers
  GET /getAllProducts
  POST /createOrder

2. 集合使用复数名词

Good:
  GET /users          # Collection
  GET /users/123      # Individual resource

Bad:
  GET /user
  GET /user/123

3. 使用小写字母和连字符

Good:
  /user-profiles
  /order-items
  /payment-methods

Bad:
  /userProfiles
  /OrderItems
  /payment_methods

4. 相关资源采用层级结构

Good:
  /users/123/posts
  /users/123/posts/456
  /users/123/posts/456/comments

Avoid Deep Nesting (max 2-3 levels):
  /organizations/1/departments/2/teams/3/members/4/tasks/5  # Too deep!

资源设计模式

模式 1：集合和条目资源

Collection Resource:
  GET    /products          # List all products
  POST   /products          # Create new product

Item Resource:
  GET    /products/123      # Get specific product
  PUT    /products/123      # Replace product (full update)
  PATCH  /products/123      # Partial update
  DELETE /products/123      # Delete product

模式 2：嵌套资源（父子关系）

# Comments belong to posts
GET    /posts/42/comments       # List comments for post 42
POST   /posts/42/comments       # Create comment on post 42
GET    /posts/42/comments/7     # Get specific comment
DELETE /posts/42/comments/7     # Delete specific comment

# Alternative for accessing comments directly
GET    /comments/7              # Get comment by ID (if you have it)

模式 3：筛选集合（查询参数）

GET /products?category=electronics
GET /products?price_min=100&price_max=500
GET /products?sort=price&order=desc
GET /users?status=active&role=admin
GET /posts?author=123&published=true

模式 4：对资源执行操作（控制器）

对于不适合标准 CRUD 的操作：

POST /users/123/activate          # Activate user account
POST /orders/456/cancel           # Cancel order
POST /payments/789/refund         # Refund payment
POST /documents/321/publish       # Publish document
POST /subscriptions/654/renew     # Renew subscription

模式 5：批量操作

POST /users/bulk-create           # Create multiple users
PATCH /products/bulk-update       # Update multiple products
DELETE /orders/bulk-delete        # Delete multiple orders

# Or using query parameters
DELETE /orders?ids=1,2,3,4,5

深入了解 HTTP 方法

GET - 获取资源

特征：

• 安全：没有副作用
• 幂等：多次发送相同请求，效果相同
• 可缓存：响应可以缓存

FastAPI 示例：

from fastapi import FastAPI, HTTPException
from typing import List, Optional

app = FastAPI()

# Collection endpoint
@app.get("/items/")
async def list_items(
    skip: int = 0,
    limit: int = 10,
    category: Optional[str] = None
) -> List[dict]:
    """List items with pagination and filtering."""
    # Filter and paginate
    items = get_items_from_db(skip=skip, limit=limit, category=category)
    return items

# Individual resource endpoint
@app.get("/items/{item_id}")
async def get_item(item_id: int) -> dict:
    """Get a specific item by ID."""
    item = get_item_from_db(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

Express.js 示例：

const express = require('express');
const app = express();

// Collection endpoint
app.get('/items', async (req, res) => {
  try {
    const { skip = 0, limit = 10, category } = req.query;
    const items = await getItemsFromDB({ skip, limit, category });
    res.json(items);
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

// Individual resource endpoint
app.get('/items/:id', async (req, res) => {
  try {
    const item = await getItemFromDB(req.params.id);
    if (!item) {
      return res.status(404).json({ error: 'Item not found' });
    }
    res.json(item);
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

POST - 创建资源

特征：

• 不安全：有副作用（创建资源）
• 非幂等：多次发送请求会创建多个资源
• 响应应包含 Location 标头，其中列出新资源的 URI

FastAPI 示例：

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    price: float
    category: str
    description: Optional[str] = None

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemCreate, response: Response) -> dict:
    """Create a new item."""
    # Validate and create
    new_item = create_item_in_db(item)

    # Set Location header
    response.headers["Location"] = f"/items/{new_item.id}"

    return new_item

Express.js 示例：

app.use(express.json());

app.post('/items', async (req, res) => {
  try {
    const { name, price, category, description } = req.body;

    // Validate
    if (!name || !price || !category) {
      return res.status(400).json({
        error: 'Missing required fields: name, price, category'
      });
    }

    // Create resource
    const newItem = await createItemInDB({ name, price, category, description });

    // Set Location header and return 201
    res.location(`/items/${newItem.id}`)
       .status(201)
       .json(newItem);
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

PUT - 替换资源

特征：

• 不安全：有副作用
• 幂等：多次发送相同请求，效果相同
• 替换整个资源（所有字段均为必填）

FastAPI 示例：

class ItemUpdate(BaseModel):
    name: str
    price: float
    category: str
    description: str

@app.put("/items/{item_id}")
async def replace_item(item_id: int, item: ItemUpdate) -> dict:
    """Replace an entire item (all fields required)."""
    existing_item = get_item_from_db(item_id)
    if not existing_item:
        raise HTTPException(status_code=404, detail="Item not found")

    # Replace entire resource
    updated_item = replace_item_in_db(item_id, item)
    return updated_item

Express.js 示例：

app.put('/items/:id', async (req, res) => {
  try {
    const { name, price, category, description } = req.body;

    // All fields required for PUT
    if (!name || !price || !category || description === undefined) {
      return res.status(400).json({
        error: 'PUT requires all fields: name, price, category, description'
      });
    }

    const existingItem = await getItemFromDB(req.params.id);
    if (!existingItem) {
      return res.status(404).json({ error: 'Item not found' });
    }

    // Replace entire resource
    const updatedItem = await replaceItemInDB(req.params.id, req.body);
    res.json(updatedItem);
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

PATCH - 部分更新

特征：

• 不安全：有副作用
• 幂等：多次发送相同请求，效果相同
• 仅更新指定字段（部分更新）

FastAPI 示例：

class ItemPatch(BaseModel):
    name: Optional[str] = None
    price: Optional[float] = None
    category: Optional[str] = None
    description: Optional[str] = None

@app.patch("/items/{item_id}")
async def update_item(item_id: int, item: ItemPatch) -> dict:
    """Partially update an item (only provided fields)."""
    existing_item = get_item_from_db(item_id)
    if not existing_item:
        raise HTTPException(status_code=404, detail="Item not found")

    # Update only provided fields
    update_data = item.model_dump(exclude_unset=True)
    updated_item = update_item_in_db(item_id, update_data)
    return updated_item

Express.js 示例：

app.patch('/items/:id', async (req, res) => {
  try {
    const existingItem = await getItemFromDB(req.params.id);
    if (!existingItem) {
      return res.status(404).json({ error: 'Item not found' });
    }

    // Update only provided fields
    const updatedItem = await updateItemInDB(req.params.id, req.body);
    res.json(updatedItem);
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

DELETE - 删除资源

特性：

• 不安全：会产生副作用
• 幂等：多个相同请求的效果相同
• 返回 204 No Content 或包含响应正文的 200 OK

FastAPI 示例：

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    """Delete an item."""
    existing_item = get_item_from_db(item_id)
    if not existing_item:
        raise HTTPException(status_code=404, detail="Item not found")

    delete_item_from_db(item_id)
    return None  # 204 No Content

# Alternative: Return deleted resource
@app.delete("/items/{item_id}")
async def delete_item_with_response(item_id: int) -> dict:
    """Delete an item and return it."""
    existing_item = get_item_from_db(item_id)
    if not existing_item:
        raise HTTPException(status_code=404, detail="Item not found")

    delete_item_from_db(item_id)
    return existing_item  # 200 OK with body

Express.js 示例：

// 204 No Content approach
app.delete('/items/:id', async (req, res) => {
  try {
    const existingItem = await getItemFromDB(req.params.id);
    if (!existingItem) {
      return res.status(404).json({ error: 'Item not found' });
    }

    await deleteItemFromDB(req.params.id);
    res.status(204).send();
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

// 200 OK with response body approach
app.delete('/items/:id', async (req, res) => {
  try {
    const existingItem = await getItemFromDB(req.params.id);
    if (!existingItem) {
      return res.status(404).json({ error: 'Item not found' });
    }

    await deleteItemFromDB(req.params.id);
    res.json({ message: 'Item deleted successfully', item: existingItem });
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

API 版本控制策略

策略 1：URI 版本控制（最常见）

在 URI 路径中指定版本——清晰、明确且易于理解。

优点：

• 明确且直观
• 易于路由到不同版本的代码
• 适合浏览器使用
• 易于编写文档

缺点：

• 会创建多个端点
• 可能导致代码重复
• 不同版本之间 URL 会发生变化

FastAPI 实现：

from fastapi import FastAPI, APIRouter

app = FastAPI()

# Version 1 router
v1_router = APIRouter(prefix="/api/v1")

@v1_router.get("/users")
async def get_users_v1():
    return {"users": ["user1", "user2"], "version": "1.0"}

@v1_router.get("/users/{user_id}")
async def get_user_v1(user_id: int):
    return {"id": user_id, "name": "John", "version": "1.0"}

# Version 2 router
v2_router = APIRouter(prefix="/api/v2")

@v2_router.get("/users")
async def get_users_v2(limit: int = 10, offset: int = 0):
    """V2 adds pagination"""
    return {
        "users": ["user1", "user2"],
        "pagination": {"limit": limit, "offset": offset},
        "version": "2.0"
    }

@v2_router.get("/users/{user_id}")
async def get_user_v2(user_id: int):
    """V2 returns more fields"""
    return {
        "id": user_id,
        "name": "John",
        "email": "[email protected]",
        "created_at": "2024-01-01",
        "version": "2.0"
    }

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

Express.js 实现：

const express = require('express');
const app = express();

// Version 1 routes
const v1Router = express.Router();

v1Router.get('/users', (req, res) => {
  res.json({ users: ['user1', 'user2'], version: '1.0' });
});

v1Router.get('/users/:id', (req, res) => {
  res.json({ id: req.params.id, name: 'John', version: '1.0' });
});

// Version 2 routes
const v2Router = express.Router();

v2Router.get('/users', (req, res) => {
  const { limit = 10, offset = 0 } = req.query;
  res.json({
    users: ['user1', 'user2'],
    pagination: { limit, offset },
    version: '2.0'
  });
});

v2Router.get('/users/:id', (req, res) => {
  res.json({
    id: req.params.id,
    name: 'John',
    email: '[email protected]',
    created_at: '2024-01-01',
    version: '2.0'
  });
});

app.use('/api/v1', v1Router);
app.use('/api/v2', v2Router);

策略 2：标头版本控制

在自定义标头或 Accept 标头中指定版本。

优点：

• URI 简洁
• 不会污染 URL
• 更符合“RESTful”风格（资源只有一个 URI）

缺点：

• 不够直观
• 在浏览器中测试更困难
• 路由逻辑更复杂

FastAPI 实现：

from fastapi import FastAPI, Header, HTTPException

app = FastAPI()

@app.get("/users")
async def get_users(api_version: str = Header(default="1.0", alias="X-API-Version")):
    """Handle multiple versions based on header"""
    if api_version == "1.0":
        return {"users": ["user1", "user2"], "version": "1.0"}
    elif api_version == "2.0":
        return {
            "users": ["user1", "user2"],
            "pagination": {"limit": 10, "offset": 0},
            "version": "2.0"
        }
    else:
        raise HTTPException(
            status_code=400,
            detail=f"Unsupported API version: {api_version}"
        )

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    api_version: str = Header(default="1.0", alias="X-API-Version")
):
    """User endpoint with version handling"""
    if api_version == "1.0":
        return {"id": user_id, "name": "John", "version": "1.0"}
    elif api_version == "2.0":
        return {
            "id": user_id,
            "name": "John",
            "email": "[email protected]",
            "created_at": "2024-01-01",
            "version": "2.0"
        }
    else:
        raise HTTPException(
            status_code=400,
            detail=f"Unsupported API version: {api_version}"
        )

Express.js 实现：

app.get('/users', (req, res) => {
  const version = req.get('X-API-Version') || '1.0';

  if (version === '1.0') {
    res.json({ users: ['user1', 'user2'], version: '1.0' });
  } else if (version === '2.0') {
    res.json({
      users: ['user1', 'user2'],
      pagination: { limit: 10, offset: 0 },
      version: '2.0'
    });
  } else {
    res.status(400).json({ error: `Unsupported API version: ${version}` });
  }
});

app.get('/users/:id', (req, res) => {
  const version = req.get('X-API-Version') || '1.0';

  if (version === '1.0') {
    res.json({ id: req.params.id, name: 'John', version: '1.0' });
  } else if (version === '2.0') {
    res.json({
      id: req.params.id,
      name: 'John',
      email: '[email protected]',
      created_at: '2024-01-01',
      version: '2.0'
    });
  } else {
    res.status(400).json({ error: `Unsupported API version: ${version}` });
  }
});

策略 3：内容协商（Accept 标头）

在 Accept 标头中使用自定义媒体类型指定版本。

FastAPI 实现：

from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

@app.get("/users")
async def get_users(request: Request):
    """Handle versioning via Accept header"""
    accept = request.headers.get("accept", "application/vnd.api.v1+json")

    if "vnd.api.v1+json" in accept:
        return {"users": ["user1", "user2"], "version": "1.0"}
    elif "vnd.api.v2+json" in accept:
        return {
            "users": ["user1", "user2"],
            "pagination": {"limit": 10, "offset": 0},
            "version": "2.0"
        }
    else:
        raise HTTPException(
            status_code=406,
            detail="Not Acceptable: Unsupported media type"
        )

Express.js 实现：

app.get('/users', (req, res) => {
  const accept = req.get('Accept') || 'application/vnd.api.v1+json';

  if (accept.includes('vnd.api.v1+json')) {
    res.type('application/vnd.api.v1+json')
       .json({ users: ['user1', 'user2'], version: '1.0' });
  } else if (accept.includes('vnd.api.v2+json')) {
    res.type('application/vnd.api.v2+json')
       .json({
         users: ['user1', 'user2'],
         pagination: { limit: 10, offset: 0 },
         version: '2.0'
       });
  } else {
    res.status(406).json({ error: 'Not Acceptable: Unsupported media type' });
  }
});

策略 4：查询参数版本控制

将版本指定为查询参数（最不推荐）。

GET /users?version=2.0
GET /users/123?v=2

缺点：

• 将版本控制与筛选混在一起
• 更难缓存
• 关注点分离不够清晰

分页模式

模式 1：基于偏移量的分页（传统方式）

实现简单，但处理大型数据集时可能出现性能问题。

FastAPI 实现：

from fastapi import FastAPI, Query
from typing import List
from pydantic import BaseModel

class PaginatedResponse(BaseModel):
    items: List[dict]
    total: int
    limit: int
    offset: int
    has_more: bool

@app.get("/items", response_model=PaginatedResponse)
async def list_items(
    limit: int = Query(default=10, ge=1, le=100),
    offset: int = Query(default=0, ge=0)
):
    """Offset-based pagination"""
    # Get total count
    total = count_items_in_db()

    # Get paginated items
    items = get_items_from_db(limit=limit, offset=offset)

    # Check if there are more items
    has_more = (offset + limit) < total

    return {
        "items": items,
        "total": total,
        "limit": limit,
        "offset": offset,
        "has_more": has_more
    }

Express.js 实现：

app.get('/items', async (req, res) => {
  try {
    const limit = Math.min(parseInt(req.query.limit) || 10, 100);
    const offset = parseInt(req.query.offset) || 0;

    const total = await countItemsInDB();
    const items = await getItemsFromDB({ limit, offset });
    const hasMore = (offset + limit) < total;

    res.json({
      items,
      total,
      limit,
      offset,
      has_more: hasMore
    });
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

模式 2：基于游标的分页（大型数据集推荐）

处理大型数据集时效率更高，还能避免分页期间数据变更带来的问题。

FastAPI 实现：

from typing import Optional

class CursorPaginatedResponse(BaseModel):
    items: List[dict]
    next_cursor: Optional[str] = None
    has_more: bool

@app.get("/items", response_model=CursorPaginatedResponse)
async def list_items_cursor(
    limit: int = Query(default=10, ge=1, le=100),
    cursor: Optional[str] = None
):
    """Cursor-based pagination"""
    # Get items after cursor
    items = get_items_after_cursor(cursor=cursor, limit=limit + 1)

    # Check if there are more items
    has_more = len(items) > limit

    # Get next cursor from last item
    next_cursor = None
    if has_more:
        items = items[:limit]  # Remove extra item
        next_cursor = items[-1]["id"]  # Use last item ID as cursor

    return {
        "items": items,
        "next_cursor": next_cursor,
        "has_more": has_more
    }

Express.js 实现：

app.get('/items', async (req, res) => {
  try {
    const limit = Math.min(parseInt(req.query.limit) || 10, 100);
    const cursor = req.query.cursor || null;

    // Get one extra item to check if there are more
    const items = await getItemsAfterCursor({ cursor, limit: limit + 1 });

    const hasMore = items.length > limit;
    let nextCursor = null;

    if (hasMore) {
      items.pop(); // Remove extra item
      nextCursor = items[items.length - 1].id; // Last item ID as cursor
    }

    res.json({
      items,
      next_cursor: nextCursor,
      has_more: hasMore
    });
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

模式 3：基于页码的分页

适用于带页码的 UI，方便用户使用。

FastAPI 实现：

class PagePaginatedResponse(BaseModel):
    items: List[dict]
    page: int
    page_size: int
    total_pages: int
    total_items: int

@app.get("/items", response_model=PagePaginatedResponse)
async def list_items_pages(
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=10, ge=1, le=100)
):
    """Page-based pagination"""
    # Calculate offset
    offset = (page - 1) * page_size

    # Get total count and items
    total_items = count_items_in_db()
    items = get_items_from_db(limit=page_size, offset=offset)

    # Calculate total pages
    total_pages = (total_items + page_size - 1) // page_size

    return {
        "items": items,
        "page": page,
        "page_size": page_size,
        "total_pages": total_pages,
        "total_items": total_items
    }

Express.js 实现：

app.get('/items', async (req, res) => {
  try {
    const page = Math.max(parseInt(req.query.page) || 1, 1);
    const pageSize = Math.min(parseInt(req.query.page_size) || 10, 100);

    const offset = (page - 1) * pageSize;

    const totalItems = await countItemsInDB();
    const items = await getItemsFromDB({ limit: pageSize, offset });
    const totalPages = Math.ceil(totalItems / pageSize);

    res.json({
      items,
      page,
      page_size: pageSize,
      total_pages: totalPages,
      total_items: totalItems
    });
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

筛选与排序

高级筛选

FastAPI 实现：

from enum import Enum
from typing import Optional, List

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

@app.get("/products")
async def list_products(
    # Filtering
    category: Optional[str] = None,
    min_price: Optional[float] = None,
    max_price: Optional[float] = None,
    in_stock: Optional[bool] = None,
    tags: Optional[List[str]] = Query(None),
    search: Optional[str] = None,
    # Sorting
    sort_by: Optional[str] = Query(default="created_at"),
    order: SortOrder = SortOrder.desc,
    # Pagination
    limit: int = Query(default=10, ge=1, le=100),
    offset: int = Query(default=0, ge=0)
):
    """Advanced filtering and sorting"""
    filters = {
        "category": category,
        "min_price": min_price,
        "max_price": max_price,
        "in_stock": in_stock,
        "tags": tags,
        "search": search
    }

    # Remove None values
    filters = {k: v for k, v in filters.items() if v is not None}

    # Query database
    products = query_products(
        filters=filters,
        sort_by=sort_by,
        order=order.value,
        limit=limit,
        offset=offset
    )

    return {
        "products": products,
        "filters": filters,
        "sort": {"by": sort_by, "order": order.value},
        "pagination": {"limit": limit, "offset": offset}
    }

Express.js 实现：

app.get('/products', async (req, res) => {
  try {
    const {
      category,
      min_price,
      max_price,
      in_stock,
      tags,
      search,
      sort_by = 'created_at',
      order = 'desc',
      limit = 10,
      offset = 0
    } = req.query;

    // Build filters
    const filters = {};
    if (category) filters.category = category;
    if (min_price) filters.min_price = parseFloat(min_price);
    if (max_price) filters.max_price = parseFloat(max_price);
    if (in_stock !== undefined) filters.in_stock = in_stock === 'true';
    if (tags) filters.tags = Array.isArray(tags) ? tags : [tags];
    if (search) filters.search = search;

    // Query database
    const products = await queryProducts({
      filters,
      sortBy: sort_by,
      order,
      limit: Math.min(parseInt(limit), 100),
      offset: parseInt(offset)
    });

    res.json({
      products,
      filters,
      sort: { by: sort_by, order },
      pagination: { limit, offset }
    });
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

错误处理最佳实践

一致的错误响应格式

标准错误响应结构：

{
  "error": {
    "code": "RESOURCE_NOT_FOUND",
    "message": "The requested user was not found",
    "details": {
      "user_id": "123",
      "resource": "user"
    },
    "timestamp": "2024-01-15T10:30:00Z"
  }
}

HTTP 状态码

成功状态码：

• 200 OK：GET、PUT、PATCH、DELETE 请求成功，且包含响应
• 201 Created：POST 请求成功并创建了资源
• 202 Accepted：已接受请求，进行异步处理
• 204 No Content：DELETE 请求成功，且没有响应正文

客户端错误状态码：

• 400 Bad Request：请求语法无效或验证错误
• 401 Unauthorized：需要身份验证
• 403 Forbidden：已通过身份验证，但未获授权
• 404 Not Found：资源不存在
• 405 Method Not Allowed：不支持该 HTTP 方法
• 409 Conflict：请求与当前状态冲突
• 422 Unprocessable Entity：验证失败
• 429 Too Many Requests：超出速率限制

服务器错误状态码：

• 500 Internal Server Error：通用服务器错误
• 502 Bad Gateway：上游服务器返回了无效响应
• 503 Service Unavailable：服务器暂时不可用
• 504 Gateway Timeout：上游服务器超时

FastAPI 错误处理

from fastapi import FastAPI, HTTPException, status, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from datetime import datetime

app = FastAPI()

# Custom exception
class ResourceNotFoundError(Exception):
    def __init__(self, resource: str, resource_id: str):
        self.resource = resource
        self.resource_id = resource_id

# Global exception handler
@app.exception_handler(ResourceNotFoundError)
async def resource_not_found_handler(request: Request, exc: ResourceNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "error": {
                "code": "RESOURCE_NOT_FOUND",
                "message": f"The requested {exc.resource} was not found",
                "details": {
                    "resource": exc.resource,
                    "resource_id": exc.resource_id
                },
                "timestamp": datetime.utcnow().isoformat() + "Z"
            }
        }
    )

# Validation error handler
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "error": {
                "code": "VALIDATION_ERROR",
                "message": "Request validation failed",
                "details": exc.errors(),
                "timestamp": datetime.utcnow().isoformat() + "Z"
            }
        }
    )

# Using exceptions
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = get_user_from_db(user_id)
    if not user:
        raise ResourceNotFoundError("user", str(user_id))
    return user

# Manual error responses
@app.post("/users")
async def create_user(email: str):
    if user_exists(email):
        raise HTTPException(
            status_code=409,
            detail={
                "code": "DUPLICATE_EMAIL",
                "message": "A user with this email already exists",
                "details": {"email": email}
            }
        )
    return create_user_in_db(email)

Express.js 错误处理

const express = require('express');
const app = express();

app.use(express.json());

// Custom error class
class ResourceNotFoundError extends Error {
  constructor(resource, resourceId) {
    super(`${resource} not found`);
    this.name = 'ResourceNotFoundError';
    this.resource = resource;
    this.resourceId = resourceId;
    this.statusCode = 404;
  }
}

class ValidationError extends Error {
  constructor(message, details) {
    super(message);
    this.name = 'ValidationError';
    this.details = details;
    this.statusCode = 422;
  }
}

// Routes
app.get('/users/:id', async (req, res, next) => {
  try {
    const user = await getUserFromDB(req.params.id);
    if (!user) {
      throw new ResourceNotFoundError('user', req.params.id);
    }
    res.json(user);
  } catch (error) {
    next(error);
  }
});

app.post('/users', async (req, res, next) => {
  try {
    const { email } = req.body;

    if (!email) {
      throw new ValidationError('Validation failed', {
        field: 'email',
        message: 'Email is required'
      });
    }

    const userExists = await checkUserExists(email);
    if (userExists) {
      const error = new Error('Duplicate email');
      error.statusCode = 409;
      error.code = 'DUPLICATE_EMAIL';
      error.details = { email };
      throw error;
    }

    const newUser = await createUserInDB(email);
    res.status(201).json(newUser);
  } catch (error) {
    next(error);
  }
});

// Global error handler (must be last)
app.use((err, req, res, next) => {
  const statusCode = err.statusCode || 500;
  const code = err.code || err.name || 'INTERNAL_SERVER_ERROR';

  const errorResponse = {
    error: {
      code,
      message: err.message,
      timestamp: new Date().toISOString()
    }
  };

  // Add details if available
  if (err.details) {
    errorResponse.error.details = err.details;
  } else if (err.resource && err.resourceId) {
    errorResponse.error.details = {
      resource: err.resource,
      resource_id: err.resourceId
    };
  }

  // Log error for debugging (don't expose in production)
  if (process.env.NODE_ENV !== 'production') {
    errorResponse.error.stack = err.stack;
  }

  res.status(statusCode).json(errorResponse);
});

HATEOAS 与超媒体

什么是 HATEOAS？

HATEOAS（超媒体作为应用状态引擎）是指在 API 响应中包含指向相关资源的链接。

优点：

• API 可自我说明
• 客户端无需构造 URL
• 更易演进 API
• 更易发现资源

FastAPI HATEOAS 实现

from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl
from typing import List, Optional

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

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    links: List[Link]

class UserListResponse(BaseModel):
    users: List[UserResponse]
    links: List[Link]

app = FastAPI()

def build_user_links(user_id: int, base_url: str = "http://api.example.com") -> List[Link]:
    """Build HATEOAS links for a user"""
    return [
        Link(rel="self", href=f"{base_url}/users/{user_id}", method="GET"),
        Link(rel="update", href=f"{base_url}/users/{user_id}", method="PUT"),
        Link(rel="delete", href=f"{base_url}/users/{user_id}", method="DELETE"),
        Link(rel="posts", href=f"{base_url}/users/{user_id}/posts", method="GET"),
        Link(rel="create_post", href=f"{base_url}/users/{user_id}/posts", method="POST")
    ]

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    """Get user with HATEOAS links"""
    user = get_user_from_db(user_id)

    return {
        "id": user.id,
        "name": user.name,
        "email": user.email,
        "links": build_user_links(user.id)
    }

@app.get("/users", response_model=UserListResponse)
async def list_users():
    """List users with HATEOAS links"""
    users = get_users_from_db()

    users_with_links = [
        {
            "id": user.id,
            "name": user.name,
            "email": user.email,
            "links": build_user_links(user.id)
        }
        for user in users
    ]

    collection_links = [
        Link(rel="self", href="http://api.example.com/users", method="GET"),
        Link(rel="create", href="http://api.example.com/users", method="POST")
    ]

    return {
        "users": users_with_links,
        "links": collection_links
    }

Express.js HATEOAS 实现

app.get('/users/:id', async (req, res) => {
  try {
    const user = await getUserFromDB(req.params.id);
    if (!user) {
      return res.status(404).json({ error: 'User not found' });
    }

    const baseUrl = `${req.protocol}://${req.get('host')}`;

    res.json({
      id: user.id,
      name: user.name,
      email: user.email,
      _links: {
        self: { href: `${baseUrl}/users/${user.id}`, method: 'GET' },
        update: { href: `${baseUrl}/users/${user.id}`, method: 'PUT' },
        delete: { href: `${baseUrl}/users/${user.id}`, method: 'DELETE' },
        posts: { href: `${baseUrl}/users/${user.id}/posts`, method: 'GET' },
        create_post: { href: `${baseUrl}/users/${user.id}/posts`, method: 'POST' }
      }
    });
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

app.get('/users', async (req, res) => {
  try {
    const users = await getUsersFromDB();
    const baseUrl = `${req.protocol}://${req.get('host')}`;

    const usersWithLinks = users.map(user => ({
      id: user.id,
      name: user.name,
      email: user.email,
      _links: {
        self: { href: `${baseUrl}/users/${user.id}`, method: 'GET' },
        update: { href: `${baseUrl}/users/${user.id}`, method: 'PUT' },
        delete: { href: `${baseUrl}/users/${user.id}`, method: 'DELETE' }
      }
    }));

    res.json({
      users: usersWithLinks,
      _links: {
        self: { href: `${baseUrl}/users`, method: 'GET' },
        create: { href: `${baseUrl}/users`, method: 'POST' }
      }
    });
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

性能优化

使用 ETag 缓存

FastAPI 实现：

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
import hashlib

app = FastAPI()

def generate_etag(data: dict) -> str:
    """Generate ETag from response data"""
    content = str(data).encode('utf-8')
    return hashlib.md5(content).hexdigest()

@app.get("/users/{user_id}")
async def get_user_cached(user_id: int, request: Request):
    """Get user with ETag caching"""
    user = get_user_from_db(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    # Generate ETag
    etag = generate_etag(user)

    # Check If-None-Match header
    if_none_match = request.headers.get("if-none-match")
    if if_none_match == etag:
        return Response(status_code=304)  # Not Modified

    # Return with ETag header
    return JSONResponse(
        content=user,
        headers={"ETag": etag, "Cache-Control": "max-age=300"}
    )

Express.js 实现：

const crypto = require('crypto');

function generateETag(data) {
  const content = JSON.stringify(data);
  return crypto.createHash('md5').update(content).digest('hex');
}

app.get('/users/:id', async (req, res) => {
  try {
    const user = await getUserFromDB(req.params.id);
    if (!user) {
      return res.status(404).json({ error: 'User not found' });
    }

    const etag = generateETag(user);

    // Check If-None-Match header
    if (req.get('If-None-Match') === etag) {
      return res.status(304).send(); // Not Modified
    }

    res.set('ETag', etag)
       .set('Cache-Control', 'max-age=300')
       .json(user);
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
});

速率限制

Express.js 实现：

const rateLimit = require('express-rate-limit');

// Create rate limiter
const apiLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15 minutes
  max: 100, // Limit each IP to 100 requests per windowMs
  message: {
    error: {
      code: 'RATE_LIMIT_EXCEEDED',
      message: 'Too many requests, please try again later'
    }
  },
  standardHeaders: true, // Return rate limit info in headers
  legacyHeaders: false
});

// Apply to all routes
app.use('/api/', apiLimiter);

// Or create specific limiters
const createAccountLimiter = rateLimit({
  windowMs: 60 * 60 * 1000, // 1 hour
  max: 5, // 5 requests per hour
  message: 'Too many accounts created, please try again later'
});

app.post('/api/users', createAccountLimiter, async (req, res) => {
  // Create user
});

压缩

FastAPI 实现：

from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

# Add compression middleware
app.add_middleware(GZipMiddleware, minimum_size=1000)

@app.get("/large-data")
async def get_large_data():
    """This response will be compressed if > 1000 bytes"""
    return {"data": [{"id": i, "value": f"item_{i}"} for i in range(1000)]}

Express.js 实现：

const compression = require('compression');

// Add compression middleware
app.use(compression({
  threshold: 1024, // Only compress responses > 1KB
  level: 6 // Compression level (0-9)
}));

app.get('/large-data', (req, res) => {
  const data = Array.from({ length: 1000 }, (_, i) => ({
    id: i,
    value: `item_${i}`
  }));
  res.json({ data });
});

安全最佳实践

身份验证模式

FastAPI 中的 JWT 身份验证：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from datetime import datetime, timedelta

app = FastAPI()
security = HTTPBearer()

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: timedelta = None):
    """Create JWT token"""
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """Verify JWT token"""
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token has expired"
        )
    except jwt.JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials"
        )

@app.post("/login")
async def login(email: str, password: str):
    """Login and get access token"""
    user = authenticate_user(email, password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )

    access_token = create_access_token(
        data={"sub": user.email, "user_id": user.id}
    )

    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me")
async def get_current_user(payload: dict = Depends(verify_token)):
    """Protected endpoint - requires authentication"""
    user_id = payload.get("user_id")
    user = get_user_from_db(user_id)
    return user

Express.js 中的 JWT 身份验证：

const jwt = require('jsonwebtoken');
const express = require('express');
const app = express();

const SECRET_KEY = 'your-secret-key';

function createAccessToken(data, expiresIn = '15m') {
  return jwt.sign(data, SECRET_KEY, { expiresIn });
}

function verifyToken(req, res, next) {
  const authHeader = req.get('Authorization');

  if (!authHeader || !authHeader.startsWith('Bearer ')) {
    return res.status(401).json({ error: 'Missing or invalid authorization header' });
  }

  const token = authHeader.substring(7);

  try {
    const payload = jwt.verify(token, SECRET_KEY);
    req.user = payload;
    next();
  } catch (error) {
    if (error.name === 'TokenExpiredError') {
      return res.status(401).json({ error: 'Token has expired' });
    }
    return res.status(401).json({ error: 'Invalid token' });
  }
}

app.post('/login', async (req, res) => {
  const { email, password } = req.body;

  const user = await authenticateUser(email, password);
  if (!user) {
    return res.status(401).json({ error: 'Incorrect email or password' });
  }

  const accessToken = createAccessToken({
    sub: user.email,
    user_id: user.id
  });

  res.json({ access_token: accessToken, token_type: 'bearer' });
});

app.get('/users/me', verifyToken, async (req, res) => {
  const user = await getUserFromDB(req.user.user_id);
  res.json(user);
});

输入验证与清理

FastAPI 验证：

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional

class UserCreate(BaseModel):
    email: EmailStr
    username: str = Field(..., min_length=3, max_length=50, pattern="^[a-zA-Z0-9_-]+$")
    password: str = Field(..., min_length=8)
    age: Optional[int] = Field(None, ge=0, le=150)

    @validator('password')
    def password_strength(cls, v):
        """Validate password strength"""
        if not any(char.isdigit() for char in v):
            raise ValueError('Password must contain at least one digit')
        if not any(char.isupper() for char in v):
            raise ValueError('Password must contain at least one uppercase letter')
        return v

@app.post("/users")
async def create_user(user: UserCreate):
    """Automatically validates input"""
    # Input is already validated by Pydantic
    hashed_password = hash_password(user.password)
    new_user = create_user_in_db(user.email, user.username, hashed_password)
    return new_user

API 文档

使用 FastAPI 编写 OpenAPI/Swagger 文档

FastAPI 会自动生成 OpenAPI 文档：

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(
    title="My API",
    description="Comprehensive API for managing resources",
    version="1.0.0",
    docs_url="/api/docs",
    redoc_url="/api/redoc"
)

class Item(BaseModel):
    """Item model with rich documentation"""
    name: str = Field(..., description="The name of the item", example="Widget")
    price: float = Field(..., description="Price in USD", example=19.99, gt=0)
    description: str = Field(None, description="Optional item description")

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "name": "Super Widget",
                    "price": 29.99,
                    "description": "An amazing widget"
                }
            ]
        }
    }

@app.post(
    "/items",
    response_model=Item,
    status_code=201,
    summary="Create a new item",
    description="Create a new item with name, price, and optional description",
    response_description="The created item",
    tags=["items"]
)
async def create_item(item: Item):
    """
    Create a new item with all the information:

    - **name**: The item name (required)
    - **price**: The item price in USD (required, must be positive)
    - **description**: Optional description of the item
    """
    return item

最佳实践总结

API 设计原则

1. 资源使用名词，而非动词
2. 集合使用复数名词
3. 正确使用 HTTP 方法（GET、POST、PUT、PATCH、DELETE）
4. 使用恰当的 HTTP 状态码
5. 为 API 添加版本（推荐使用 URI 版本控制）
6. 为集合提供分页
7. 使用查询参数支持筛选和排序
8. 返回一致的错误响应
9. 使用 HATEOAS 提高可发现性
10. 使用 OpenAPI/Swagger 编写 API 文档

安全原则

1. 生产环境始终使用 HTTPS
2. 实施身份验证（JWT、OAuth、API 密钥）
3. 全面验证所有输入
4. 使用速率限制防止滥用
5. 清理输出以防止 XSS
6. 正确配置 CORS
7. 使用安全标头（CSP、X-Frame-Options 等）
8. 记录安全事件以便监控
9. 保持依赖项为最新版本
10. 切勿在响应中暴露敏感数据

性能原则

1. 使用缓存（ETag、Cache-Control 标头）
2. 为大型响应启用压缩
3. 为大型数据集使用分页
4. 优化数据库查询（索引、防止 N+1 问题）
5. 在 I/O 操作中使用 async/await
6. 实现连接池
7. 监控 API 性能（响应时间、错误率）
8. 为静态内容使用 CDN
9. 实现适当的日志记录，避免阻塞
10. 定期对 API 进行负载测试

---

技能版本：1.0.0 最后更新：2025 年 10 月 技能类别：API 设计、后端开发、REST 架构 兼容对象：FastAPI、Express.js、Node.js、Python、HTTP 框架