name: api-design-principles description: 掌握REST和GraphQL API设计原则,构建直观、可扩展和可维护的API,以取悦开发者。用于设计新API、审查API规范或建立API设计标准。
API设计原则
掌握REST和GraphQL API设计原则,构建直观、可扩展和可维护的API,以取悦开发者并经受时间的考验。
何时使用此技能
- 设计新的REST或GraphQL API
- 重构现有API以提高可用性
- 为团队建立API设计标准
- 在实现前审查API规范
- 迁移API范式(如从REST到GraphQL等)
- 创建开发者友好的API文档
- 为特定用例优化API(移动端、第三方集成等)
交互式设计过程
- 澄清需求:在最终确定设计前,确保理解领域和约束。
- 反馈循环:如果可用,使用
mcp-feedback-enhanced工具(例如ask_followup_question)询问澄清问题或呈现设计选项供审查;否则,使用标准聊天。
核心概念
1. RESTful设计原则
资源导向架构
- 资源是名词(用户、订单、产品),不是动词
- 使用HTTP方法进行操作(GET、POST、PUT、PATCH、DELETE)
- URL代表资源层次结构
- 一致的命名约定
HTTP方法语义:
GET:检索资源(幂等、安全)POST:创建新资源PUT:替换整个资源(幂等)PATCH:部分资源更新DELETE:删除资源(幂等)
2. GraphQL设计原则
模式优先开发
- 类型定义领域模型
- 查询用于读取数据
- 变异用于修改数据
- 订阅用于实时更新
查询结构:
- 客户端精确请求所需内容
- 单个端点,多个操作
- 强类型模式
- 内置内省
3. API版本控制策略
URL版本控制:
/api/v1/users
/api/v2/users
头部版本控制:
Accept: application/vnd.api+json; version=1
查询参数版本控制:
/api/users?version=1
REST API设计模式
模式1:资源集合设计
# 良好:资源导向端点
GET /api/users # 列出用户(带分页)
POST /api/users # 创建用户
GET /api/users/{id} # 获取特定用户
PUT /api/users/{id} # 替换用户
PATCH /api/users/{id} # 更新用户字段
DELETE /api/users/{id} # 删除用户
# 嵌套资源
GET /api/users/{id}/orders # 获取用户的订单
POST /api/users/{id}/orders # 为用户创建订单
# 不良:操作导向端点(避免)
POST /api/createUser
POST /api/getUserById
POST /api/deleteUser
模式2:分页和过滤
from typing import List, Optional
from pydantic import BaseModel, Field
class PaginationParams(BaseModel):
page: int = Field(1, ge=1, description="页码")
page_size: int = Field(20, ge=1, le=100, description="每页项目数")
class FilterParams(BaseModel):
status: Optional[str] = None
created_after: Optional[str] = None
search: Optional[str] = None
class PaginatedResponse(BaseModel):
items: List[dict]
total: int
page: int
page_size: int
pages: int
@property
def has_next(self) -> bool:
return self.page < self.pages
@property
def has_prev(self) -> bool:
return self.page > 1
# FastAPI端点示例
from fastapi import FastAPI, Query, Depends
app = FastAPI()
@app.get("/api/users", response_model=PaginatedResponse)
async def list_users(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
status: Optional[str] = Query(None),
search: Optional[str] = Query(None)
):
# 应用过滤器
query = build_query(status=status, search=search)
# 计数总数
total = await count_users(query)
# 获取页面
offset = (page - 1) * page_size
users = await fetch_users(query, limit=page_size, offset=offset)
return PaginatedResponse(
items=users,
total=total,
page=page,
page_size=page_size,
pages=(total + page_size - 1) // page_size
)
模式3:错误处理和状态码
from fastapi import HTTPException, status
from pydantic import BaseModel
class ErrorResponse(BaseModel):
error: str
message: str
details: Optional[dict] = None
timestamp: str
path: str
class ValidationErrorDetail(BaseModel):
field: str
message: str
value: Any
# 一致的错误响应
STATUS_CODES = {
"success": 200,
"created": 201,
"no_content": 204,
"bad_request": 400,
"unauthorized": 401,
"forbidden": 403,
"not_found": 404,
"conflict": 409,
"unprocessable": 422,
"internal_error": 500
}
def raise_not_found(resource: str, id: str):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={
"error": "NotFound",
"message": f"{resource} 未找到",
"details": {"id": id}
}
)
def raise_validation_error(errors: List[ValidationErrorDetail]):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail={
"error": "ValidationError",
"message": "请求验证失败",
"details": {"errors": [e.dict() for e in errors]}
}
)
# 示例用法
@app.get("/api/users/{user_id}")
async def get_user(user_id: str):
user = await fetch_user(user_id)
if not user:
raise_not_found("用户", user_id)
return user
模式4:HATEOAS(超媒体作为应用状态引擎)
class UserResponse(BaseModel):
id: str
name: str
email: str
_links: dict
@classmethod
def from_user(cls, user: User, base_url: str):
return cls(
id=user.id,
name=user.name,
email=user.email,
_links={
"self": {"href": f"{base_url}/api/users/{user.id}"},
"orders": {"href": f"{base_url}/api/users/{user.id}/orders"},
"update": {
"href": f"{base_url}/api/users/{user.id}",
"method": "PATCH"
},
"delete": {
"href": f"{base_url}/api/users/{user.id}",
"method": "DELETE"
}
}
)
GraphQL设计模式
模式1:模式设计
# schema.graphql
# 清晰的类型定义
type User {
id: ID!
email: String!
name: String!
createdAt: DateTime!
# 关系
orders(first: Int = 20, after: String, status: OrderStatus): OrderConnection!
profile: UserProfile
}
type Order {
id: ID!
status: OrderStatus!
total: Money!
items: [OrderItem!]!
createdAt: DateTime!
# 反向引用
user: User!
}
# 分页模式(Relay风格)
type OrderConnection {
edges: [OrderEdge!]!
pageInfo: PageInfo!
totalCount: Int!
}
type OrderEdge {
node: Order!
cursor: String!
}
type PageInfo {
hasNextPage: Boolean!
hasPreviousPage: Boolean!
startCursor: String
endCursor: String
}
# 枚举用于类型安全
enum OrderStatus {
PENDING
CONFIRMED
SHIPPED
DELIVERED
CANCELLED
}
# 自定义标量
scalar DateTime
scalar Money
# 查询根
type Query {
user(id: ID!): User
users(first: Int = 20, after: String, search: String): UserConnection!
order(id: ID!): Order
}
# 变异根
type Mutation {
createUser(input: CreateUserInput!): CreateUserPayload!
updateUser(input: UpdateUserInput!): UpdateUserPayload!
deleteUser(id: ID!): DeleteUserPayload!
createOrder(input: CreateOrderInput!): CreateOrderPayload!
}
# 输入类型用于变异
input CreateUserInput {
email: String!
name: String!
password: String!
}
# 负载类型用于变异
type CreateUserPayload {
user: User
errors: [Error!]
}
type Error {
field: String
message: String!
}
模式2:解析器设计
from typing import Optional, List
from ariadne import QueryType, MutationType, ObjectType
from dataclasses import dataclass
query = QueryType()
mutation = MutationType()
user_type = ObjectType("User")
@query.field("user")
async def resolve_user(obj, info, id: str) -> Optional[dict]:
"""通过ID解析单个用户。"""
return await fetch_user_by_id(id)
@query.field("users")
async def resolve_users(
obj,
info,
first: int = 20,
after: Optional[str] = None,
search: Optional[str] = None
) -> dict:
"""解析分页用户列表。"""
# 解码游标
offset = decode_cursor(after) if after else 0
# 获取用户
users = await fetch_users(
limit=first + 1, # 多获取一个以检查hasNextPage
offset=offset,
search=search
)
# 分页
has_next = len(users) > first
if has_next:
users = users[:first]
edges = [
{
"node": user,
"cursor": encode_cursor(offset + i)
}
for i, user in enumerate(users)
]
return {
"edges": edges,
"pageInfo": {
"hasNextPage": has_next,
"hasPreviousPage": offset > 0,
"startCursor": edges[0]["cursor"] if edges else None,
"endCursor": edges[-1]["cursor"] if edges else None
},
"totalCount": await count_users(search=search)
}
@user_type.field("orders")
async def resolve_user_orders(user: dict, info, first: int = 20) -> dict:
"""解析用户的订单(使用DataLoader防止N+1问题)。"""
# 使用DataLoader批量请求
loader = info.context["loaders"]["orders_by_user"]
orders = await loader.load(user["id"])
return paginate_orders(orders, first)
@mutation.field("createUser")
async def resolve_create_user(obj, info, input: dict) -> dict:
"""创建新用户。"""
try:
# 验证输入
validate_user_input(input)
# 创建用户
user = await create_user(
email=input["email"],
name=input["name"],
password=hash_password(input["password"])
)
return {
"user": user,
"errors": []
}
except ValidationError as e:
return {
"user": None,
"errors": [{"field": e.field, "message": e.message}]
}
模式3:DataLoader(防止N+1问题)
from aiodataloader import DataLoader
from typing import List, Optional
class UserLoader(DataLoader):
"""通过ID批量加载用户。"""
async def batch_load_fn(self, user_ids: List[str]) -> List[Optional[dict]]:
"""在单个查询中加载多个用户。"""
users = await fetch_users_by_ids(user_ids)
# 将结果映射回输入顺序
user_map = {user["id"]: user for user in users}
return [user_map.get(user_id) for user_id in user_ids]
class OrdersByUserLoader(DataLoader):
"""通过用户ID批量加载订单。"""
async def batch_load_fn(self, user_ids: List[str]) -> List[List[dict]]:
"""在单个查询中加载多个用户的订单。"""
orders = await fetch_orders_by_user_ids(user_ids)
# 按user_id分组订单
orders_by_user = {}
for order in orders:
user_id = order["user_id"]
if user_id not in orders_by_user:
orders_by_user[user_id] = []
orders_by_user[user_id].append(order)
# 按输入顺序返回
return [orders_by_user.get(user_id, []) for user_id in user_ids]
# 上下文设置
def create_context():
return {
"loaders": {
"user": UserLoader(),
"orders_by_user": OrdersByUserLoader()
}
}
最佳实践
REST APIs
- 一致命名:对集合使用复数名词(
/users,不是/user) - 无状态:每个请求包含所有必要信息
- 正确使用HTTP状态码:2xx成功,4xx客户端错误,5xx服务器错误
- 版本控制API:从第一天开始计划重大变更
- 分页:始终对大集合进行分页
- 速率限制:用速率限制保护您的API
- 文档:使用OpenAPI/Swagger进行交互式文档
GraphQL APIs
- 模式优先:在编写解析器前设计模式
- 避免N+1:使用DataLoader进行高效数据获取
- 输入验证:在模式和解析器级别验证
- 错误处理:在变异负载中返回结构化错误
- 分页:使用基于游标的分页(Relay规范)
- 弃用:使用
@deprecated指令进行渐进迁移 - 监控:跟踪查询复杂性和执行时间
常见陷阱
- 过度获取/不足获取(REST):在GraphQL中已解决,但需要DataLoaders
- 重大变更:版本控制API或使用弃用策略
- 不一致的错误格式:标准化错误响应
- 缺少速率限制:没有限制的API易受滥用
- 文档差:未记录的API让开发者沮丧
- 忽略HTTP语义:将POST用于幂等操作违反预期
- 紧密耦合:API结构不应反映数据库模式
资源
- references/rest-best-practices.md:全面的REST API设计指南
- references/graphql-schema-design.md:GraphQL模式模式和反模式
- references/api-versioning-strategies.md:版本控制方法和迁移路径
- assets/rest-api-template.py:FastAPI REST API模板
- assets/graphql-schema-template.graphql:完整的GraphQL模式示例
- assets/api-design-checklist.md:实施前审查清单
- scripts/openapi-generator.py:从代码生成OpenAPI规范