api-design-expert

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

API Design Expert

API设计专家

Expert guidance for API design, RESTful principles, GraphQL, versioning strategies, and API best practices.

提供API设计、RESTful原则、GraphQL、版本控制策略及API最佳实践的专业指导。

Core Concepts

核心概念

API Design Principles

API设计原则

RESTful architecture
Resource-oriented design
Uniform interface
Statelessness
Cacheability
Layered system

RESTful架构
面向资源的设计
统一接口
无状态
可缓存性
分层系统

API Styles

API风格

REST (Representational State Transfer)
GraphQL
RPC (Remote Procedure Call)
WebSocket
Server-Sent Events (SSE)
gRPC

REST (Representational State Transfer)
GraphQL
RPC (Remote Procedure Call)
WebSocket
Server-Sent Events (SSE)
gRPC

Key Considerations

核心考量点

Versioning strategies
Authentication and authorization
Rate limiting
Error handling
Documentation
Backward compatibility

版本控制策略
身份认证与授权
限流
错误处理
文档
向后兼容性

REST API Design

REST API设计

python

from fastapi import FastAPI, HTTPException, Query, Path, Header
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
from enum import Enum

app = FastAPI(
    title="User Management API",
    version="1.0.0",
    description="RESTful API for user management"
)

python

from fastapi import FastAPI, HTTPException, Query, Path, Header
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
from enum import Enum

app = FastAPI(
    title="User Management API",
    version="1.0.0",
    description="RESTful API for user management"
)

Models

class UserRole(str, Enum): ADMIN = "admin" USER = "user" GUEST = "guest"

class UserCreate(BaseModel): email: str = Field(..., example="user@example.com") name: str = Field(..., min_length=1, max_length=100) role: UserRole = UserRole.USER

class UserResponse(BaseModel): id: str email: str name: str role: UserRole created_at: datetime updated_at: datetime

class Config:
    schema_extra = {
        "example": {
            "id": "123e4567-e89b-12d3-a456-426614174000",
            "email": "user@example.com",
            "name": "John Doe",
            "role": "user",
            "created_at": "2024-01-01T00:00:00Z",
            "updated_at": "2024-01-01T00:00:00Z"
        }
    }

class UserUpdate(BaseModel): name: Optional[str] = Field(None, min_length=1, max_length=100) role: Optional[UserRole] = None

class UserRole(str, Enum): ADMIN = "admin" USER = "user" GUEST = "guest"

class UserCreate(BaseModel): email: str = Field(..., example="user@example.com") name: str = Field(..., min_length=1, max_length=100) role: UserRole = UserRole.USER

class UserResponse(BaseModel): id: str email: str name: str role: UserRole created_at: datetime updated_at: datetime

class Config:
    schema_extra = {
        "example": {
            "id": "123e4567-e89b-12d3-a456-426614174000",
            "email": "user@example.com",
            "name": "John Doe",
            "role": "user",
            "created_at": "2024-01-01T00:00:00Z",
            "updated_at": "2024-01-01T00:00:00Z"
        }
    }

class UserUpdate(BaseModel): name: Optional[str] = Field(None, min_length=1, max_length=100) role: Optional[UserRole] = None

REST Endpoints following best practices

@app.get("/api/v1/users", response_model=List[UserResponse], summary="List all users", tags=["Users"]) async def list_users( page: int = Query(1, ge=1, description="Page number"), page_size: int = Query(20, ge=1, le=100, description="Items per page"), sort: str = Query("created_at", description="Sort field"), order: str = Query("desc", regex="^(asc|desc)$") ): """ Retrieve a paginated list of users.

- **page**: Page number (starts at 1)
- **page_size**: Number of items per page (1-100)
- **sort**: Field to sort by
- **order**: Sort order (asc or desc)
"""
# Implementation
return []

@app.get("/api/v1/users/{user_id}", response_model=UserResponse, summary="Get user by ID", tags=["Users"]) async def get_user( user_id: str = Path(..., description="User ID") ): """Retrieve a specific user by ID.""" # Implementation raise HTTPException(status_code=404, detail="User not found")

@app.post("/api/v1/users", response_model=UserResponse, status_code=201, summary="Create new user", tags=["Users"]) async def create_user(user: UserCreate): """Create a new user.""" # Implementation return UserResponse( id="123e4567-e89b-12d3-a456-426614174000", email=user.email, name=user.name, role=user.role, created_at=datetime.now(), updated_at=datetime.now() )

@app.patch("/api/v1/users/{user_id}", response_model=UserResponse, summary="Update user", tags=["Users"]) async def update_user( user_id: str = Path(..., description="User ID"), user: UserUpdate = None ): """Partially update a user.""" # Implementation pass

@app.delete("/api/v1/users/{user_id}", status_code=204, summary="Delete user", tags=["Users"]) async def delete_user( user_id: str = Path(..., description="User ID") ): """Delete a user.""" # Implementation pass

- **page**: Page number (starts at 1)
- **page_size**: Number of items per page (1-100)
- **sort**: Field to sort by
- **order**: Sort order (asc or desc)
"""
# Implementation
return []

Nested resources

@app.get("/api/v1/users/{user_id}/posts", summary="Get user posts", tags=["Users", "Posts"]) async def get_user_posts(user_id: str): """Retrieve all posts for a specific user.""" return []

undefined

@app.get("/api/v1/users/{user_id}/posts", summary="Get user posts", tags=["Users", "Posts"]) async def get_user_posts(user_id: str): """Retrieve all posts for a specific user.""" return []

undefined

API Versioning

API版本控制

python

from fastapi import APIRouter, Request

python

from fastapi import APIRouter, Request

URL Path Versioning (Recommended)

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

@v1_router.get("/users") async def get_users_v1(): """Version 1: Returns basic user info""" return [{"id": 1, "name": "John"}]

@v2_router.get("/users") async def get_users_v2(): """Version 2: Returns enhanced user info""" return [{"id": 1, "name": "John", "email": "john@example.com"}]

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

@v1_router.get("/users") async def get_users_v1(): """Version 1: Returns basic user info""" return [{"id": 1, "name": "John"}]

@v2_router.get("/users") async def get_users_v2(): """Version 2: Returns enhanced user info""" return [{"id": 1, "name": "John", "email": "john@example.com"}]

Header Versioning

async def version_from_header(request: Request): version = request.headers.get("API-Version", "1") return version

@app.get("/api/users") async def get_users(version: str = Depends(version_from_header)): if version == "2": return get_users_v2() return get_users_v1()

async def version_from_header(request: Request): version = request.headers.get("API-Version", "1") return version

@app.get("/api/users") async def get_users(version: str = Depends(version_from_header)): if version == "2": return get_users_v2() return get_users_v1()

Content Negotiation Versioning

@app.get("/api/users") async def get_users_content_negotiation( accept: str = Header("application/vnd.api+json; version=1") ): if "version=2" in accept: return get_users_v2() return get_users_v1()

undefined

@app.get("/api/users") async def get_users_content_negotiation( accept: str = Header("application/vnd.api+json; version=1") ): if "version=2" in accept: return get_users_v2() return get_users_v1()

undefined

Error Handling

错误处理

python

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ValidationError

class ErrorResponse(BaseModel):
    error: str
    message: str
    details: Optional[dict] = None
    timestamp: datetime
    path: str

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            error=exc.status_code,
            message=exc.detail,
            timestamp=datetime.now(),
            path=request.url.path
        ).dict()
    )

@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    return JSONResponse(
        status_code=422,
        content=ErrorResponse(
            error="validation_error",
            message="Request validation failed",
            details=exc.errors(),
            timestamp=datetime.now(),
            path=request.url.path
        ).dict()
    )

python

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ValidationError

class ErrorResponse(BaseModel):
    error: str
    message: str
    details: Optional[dict] = None
    timestamp: datetime
    path: str

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            error=exc.status_code,
            message=exc.detail,
            timestamp=datetime.now(),
            path=request.url.path
        ).dict()
    )

@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    return JSONResponse(
        status_code=422,
        content=ErrorResponse(
            error="validation_error",
            message="Request validation failed",
            details=exc.errors(),
            timestamp=datetime.now(),
            path=request.url.path
        ).dict()
    )

Custom business logic errors

class BusinessError(Exception): def init(self, message: str, code: str): self.message = message self.code = code

@app.exception_handler(BusinessError) async def business_error_handler(request: Request, exc: BusinessError): return JSONResponse( status_code=400, content=ErrorResponse( error=exc.code, message=exc.message, timestamp=datetime.now(), path=request.url.path ).dict() )

undefined

class BusinessError(Exception): def init(self, message: str, code: str): self.message = message self.code = code

undefined

Rate Limiting

限流

python

from fastapi import Request, HTTPException
from datetime import datetime, timedelta
import redis
from typing import Dict

class RateLimiter:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    async def check_rate_limit(self,
                               key: str,
                               max_requests: int,
                               window_seconds: int) -> Dict:
        """
        Token bucket algorithm for rate limiting
        """
        now = datetime.now().timestamp()
        window_key = f"rate_limit:{key}:{int(now // window_seconds)}"

        pipe = self.redis.pipeline()
        pipe.incr(window_key)
        pipe.expire(window_key, window_seconds)
        result = pipe.execute()

        request_count = result[0]

        if request_count > max_requests:
            reset_time = (int(now // window_seconds) + 1) * window_seconds
            raise HTTPException(
                status_code=429,
                detail="Rate limit exceeded",
                headers={
                    "X-RateLimit-Limit": str(max_requests),
                    "X-RateLimit-Remaining": "0",
                    "X-RateLimit-Reset": str(reset_time)
                }
            )

        return {
            "limit": max_requests,
            "remaining": max_requests - request_count,
            "reset": (int(now // window_seconds) + 1) * window_seconds
        }

python

from fastapi import Request, HTTPException
from datetime import datetime, timedelta
import redis
from typing import Dict

class RateLimiter:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    async def check_rate_limit(self,
                               key: str,
                               max_requests: int,
                               window_seconds: int) -> Dict:
        """
        Token bucket algorithm for rate limiting
        """
        now = datetime.now().timestamp()
        window_key = f"rate_limit:{key}:{int(now // window_seconds)}"

        pipe = self.redis.pipeline()
        pipe.incr(window_key)
        pipe.expire(window_key, window_seconds)
        result = pipe.execute()

        request_count = result[0]

        if request_count > max_requests:
            reset_time = (int(now // window_seconds) + 1) * window_seconds
            raise HTTPException(
                status_code=429,
                detail="Rate limit exceeded",
                headers={
                    "X-RateLimit-Limit": str(max_requests),
                    "X-RateLimit-Remaining": "0",
                    "X-RateLimit-Reset": str(reset_time)
                }
            )

        return {
            "limit": max_requests,
            "remaining": max_requests - request_count,
            "reset": (int(now // window_seconds) + 1) * window_seconds
        }

Middleware for rate limiting

@app.middleware("http") async def rate_limit_middleware(request: Request, call_next): limiter = RateLimiter(redis_client)

# Get client identifier (IP, API key, etc.)
client_id = request.client.host

try:
    rate_info = await limiter.check_rate_limit(
        client_id,
        max_requests=100,
        window_seconds=60
    )

    response = await call_next(request)

    # Add rate limit headers
    response.headers["X-RateLimit-Limit"] = str(rate_info["limit"])
    response.headers["X-RateLimit-Remaining"] = str(rate_info["remaining"])
    response.headers["X-RateLimit-Reset"] = str(rate_info["reset"])

    return response
except HTTPException as e:
    return JSONResponse(
        status_code=e.status_code,
        content={"error": e.detail},
        headers=e.headers
    )

undefined

@app.middleware("http") async def rate_limit_middleware(request: Request, call_next): limiter = RateLimiter(redis_client)

# Get client identifier (IP, API key, etc.)
client_id = request.client.host

try:
    rate_info = await limiter.check_rate_limit(
        client_id,
        max_requests=100,
        window_seconds=60
    )

    response = await call_next(request)

    # Add rate limit headers
    response.headers["X-RateLimit-Limit"] = str(rate_info["limit"])
    response.headers["X-RateLimit-Remaining"] = str(rate_info["remaining"])
    response.headers["X-RateLimit-Reset"] = str(rate_info["reset"])

    return response
except HTTPException as e:
    return JSONResponse(
        status_code=e.status_code,
        content={"error": e.detail},
        headers=e.headers
    )

undefined

HATEOAS Implementation

HATEOAS实现

python

from typing import Dict, List

class HATEOASResponse(BaseModel):
    data: dict
    links: Dict[str, str]

def create_links(resource_id: str, resource_type: str) -> Dict[str, str]:
    """Create HATEOAS links for a resource"""
    return {
        "self": f"/api/v1/{resource_type}/{resource_id}",
        "update": f"/api/v1/{resource_type}/{resource_id}",
        "delete": f"/api/v1/{resource_type}/{resource_id}",
        "collection": f"/api/v1/{resource_type}"
    }

@app.get("/api/v1/users/{user_id}", response_model=HATEOASResponse)
async def get_user_with_links(user_id: str):
    user = get_user_from_db(user_id)

    return HATEOASResponse(
        data=user,
        links={
            "self": f"/api/v1/users/{user_id}",
            "posts": f"/api/v1/users/{user_id}/posts",
            "profile": f"/api/v1/users/{user_id}/profile",
            "update": f"/api/v1/users/{user_id}",
            "delete": f"/api/v1/users/{user_id}"
        }
    )

python

from typing import Dict, List

class HATEOASResponse(BaseModel):
    data: dict
    links: Dict[str, str]

def create_links(resource_id: str, resource_type: str) -> Dict[str, str]:
    """Create HATEOAS links for a resource"""
    return {
        "self": f"/api/v1/{resource_type}/{resource_id}",
        "update": f"/api/v1/{resource_type}/{resource_id}",
        "delete": f"/api/v1/{resource_type}/{resource_id}",
        "collection": f"/api/v1/{resource_type}"
    }

@app.get("/api/v1/users/{user_id}", response_model=HATEOASResponse)
async def get_user_with_links(user_id: str):
    user = get_user_from_db(user_id)

    return HATEOASResponse(
        data=user,
        links={
            "self": f"/api/v1/users/{user_id}",
            "posts": f"/api/v1/users/{user_id}/posts",
            "profile": f"/api/v1/users/{user_id}/profile",
            "update": f"/api/v1/users/{user_id}",
            "delete": f"/api/v1/users/{user_id}"
        }
    )

API Security

API安全

python

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from passlib.context import CryptContext

security = HTTPBearer()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """Verify JWT token"""
    try:
        payload = jwt.decode(
            credentials.credentials,
            SECRET_KEY,
            algorithms=[ALGORITHM]
        )
        return payload
    except JWTError:
        raise HTTPException(
            status_code=401,
            detail="Invalid authentication credentials"
        )

@app.get("/api/v1/protected")
async def protected_route(token_payload: dict = Depends(verify_token)):
    return {"message": "Access granted", "user": token_payload}

python

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from passlib.context import CryptContext

security = HTTPBearer()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """Verify JWT token"""
    try:
        payload = jwt.decode(
            credentials.credentials,
            SECRET_KEY,
            algorithms=[ALGORITHM]
        )
        return payload
    except JWTError:
        raise HTTPException(
            status_code=401,
            detail="Invalid authentication credentials"
        )

@app.get("/api/v1/protected")
async def protected_route(token_payload: dict = Depends(verify_token)):
    return {"message": "Access granted", "user": token_payload}

API Key Authentication

async def verify_api_key(api_key: str = Header(...)): """Verify API key""" if api_key not in valid_api_keys: raise HTTPException(status_code=403, detail="Invalid API key") return api_key

undefined

async def verify_api_key(api_key: str = Header(...)): """Verify API key""" if api_key not in valid_api_keys: raise HTTPException(status_code=403, detail="Invalid API key") return api_key

undefined

Best Practices

最佳实践

Design

设计

Use nouns for resources, not verbs
Use plural names for collections
Use HTTP methods correctly (GET, POST, PUT, PATCH, DELETE)
Return appropriate status codes
Support filtering, sorting, and pagination
Use consistent naming conventions
Version APIs from the start

资源使用名词而非动词命名
集合使用复数名称
正确使用HTTP方法（GET、POST、PUT、PATCH、DELETE）
返回合适的状态码
支持过滤、排序和分页
使用统一的命名规范
从一开始就对API进行版本控制

Documentation

文档

Use OpenAPI/Swagger
Provide example requests and responses
Document error codes and messages
Include authentication requirements
Keep documentation up-to-date
Provide SDKs when possible

使用OpenAPI/Swagger
提供请求和响应示例
记录错误码和错误信息
说明身份认证要求
保持文档及时更新
尽可能提供SDK

Performance

性能

Implement caching (ETags, Cache-Control)
Support compression (gzip)
Paginate large result sets
Use async operations for long tasks
Implement rate limiting
Monitor API performance

实现缓存（ETags、Cache-Control）
支持压缩（gzip）
对大型结果集做分页
长任务使用异步操作
实现限流
监控API性能

Security

安全

Use HTTPS always
Implement authentication and authorization
Validate all inputs
Rate limit to prevent abuse
Log security events
Use API keys or OAuth 2.0
Implement CORS properly

始终使用HTTPS
实现身份认证与授权
校验所有输入
限流防止滥用
记录安全事件
使用API密钥或OAuth 2.0
正确实现CORS

Anti-Patterns

反模式

❌ Using GET for state-changing operations ❌ Returning inconsistent response formats ❌ No versioning strategy ❌ Poor error messages ❌ No rate limiting ❌ Exposing internal implementation details ❌ Breaking changes without versioning

❌ 用GET执行会更改状态的操作 ❌ 返回的响应格式不统一 ❌ 没有版本控制策略 ❌ 错误信息不清晰 ❌ 没有限流机制 ❌ 暴露内部实现细节 ❌ 不升级版本就引入破坏性变更

Resources

参考资源

REST API Design Best Practices: https://restfulapi.net/
OpenAPI Specification: https://swagger.io/specification/
API Design Guide: https://cloud.google.com/apis/design
Microsoft REST API Guidelines: https://github.com/microsoft/api-guidelines
FastAPI: https://fastapi.tiangolo.com/

REST API设计最佳实践: https://restfulapi.net/
OpenAPI规范: https://swagger.io/specification/
API设计指南: https://cloud.google.com/apis/design
Microsoft REST API规范: https://github.com/microsoft/api-guidelines
FastAPI: https://fastapi.tiangolo.com/