环境声明

  • Python 版本：Python 3.12+
  • FastAPI 版本：0.110+
  • Pydantic 版本：2.5+
  • HTTP 客户端：HTTPie 或 curl（测试用）
  • 开发工具：PyCharm 或 VS Code

学习目标

  • 深入理解 REST 架构原则与约束
  • 掌握 URL 设计与 HTTP 方法的正确用法
  • 学会设计规范的状态码与错误处理机制
  • 了解 API 版本控制的各种策略

1. REST 架构原则

1.1 什么是 REST

REST（Representational State Transfer，表述性状态转移）是一种软件架构风格，由 Roy Fielding 博士在 2000 年提出。

想象 REST 就像去餐厅点餐：

  • 资源（Resource）：菜单上的每道菜
  • 表述（Representation）：菜品的描述（文字、图片）
  • 状态转移（State Transfer）：从"未点"到"已点"到"已上菜"

1.2 REST 的六个约束

约束 说明 实践建议
客户端-服务器 分离关注点 前端专注 UI，后端专注数据
无状态 每个请求独立 请求包含所有必要信息
可缓存 响应可被缓存 合理使用缓存头
统一接口 统一的资源操作方式 使用标准 HTTP 方法
分层系统 客户端不感知中间层 使用网关、负载均衡
按需代码（可选） 服务器可下发代码 较少使用

1.3 RESTful API 的核心特征

RESTful API = 资源 + HTTP 方法 + 表述

资源：/users, /orders, /products
方法：GET, POST, PUT, DELETE, PATCH
表述：JSON, XML

2. URL 设计与 HTTP 方法

2.1 URL 设计原则

使用名词，而非动词
错误示例：
GET /getUsers
POST /createOrder
DELETE /deleteProduct/123

正确示例：
GET /users          # 获取用户列表
POST /orders        # 创建订单
DELETE /products/123 # 删除产品
使用复数形式
推荐：
GET /users
GET /orders
GET /products

不推荐：
GET /user
GET /order
GET /product
层级关系表达
# 用户的文章
GET /users/{user_id}/posts

# 文章的评论
GET /posts/{post_id}/comments

# 订单的商品
GET /orders/{order_id}/items

2.2 HTTP 方法使用规范

方法 幂等性 用途 示例
GET 是 获取资源 GET /users/123
POST 否 创建资源 POST /users
PUT 是 全量更新 PUT /users/123
PATCH 否 部分更新 PATCH /users/123
DELETE 是 删除资源 DELETE /users/123 
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI(title="RESTful API 示例")


# ==================== 数据模型 ====================

class UserBase(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None


class UserCreate(UserBase):
    password: str


class UserUpdate(BaseModel):
    email: Optional[str] = None
    full_name: Optional[str] = None


class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: str
    
    class Config:
        from_attributes = True


# 模拟数据库
users_db = {}
counter = 0


# ==================== GET 方法 ====================

@app.get("/users", response_model=List[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 100,
    is_active: Optional[bool] = None
):
    """
    获取用户列表
    
    - skip: 跳过的记录数（分页用）
    - limit: 返回的最大记录数
    - is_active: 按状态筛选
    """
    user_list = list(users_db.values())
    
    if is_active is not None:
        user_list = [u for u in user_list if u["is_active"] == is_active]
    
    return user_list[skip : skip + limit]


@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    """获取单个用户详情"""
    if user_id not in users_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"用户 {user_id} 不存在"
        )
    return users_db[user_id]


# ==================== POST 方法 ====================

@app.post(
    "/users",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED
)
async def create_user(user_data: UserCreate):
    """创建新用户"""
    global counter
    counter += 1
    
    # 检查用户名是否已存在
    for user in users_db.values():
        if user["username"] == user_data.username:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail=f"用户名 '{user_data.username}' 已存在"
            )
    
    new_user = {
        "id": counter,
        "username": user_data.username,
        "email": user_data.email,
        "full_name": user_data.full_name,
        "is_active": True,
        "created_at": "2024-01-01T00:00:00"
    }
    users_db[counter] = new_user
    
    return new_user


# ==================== PUT 方法 ====================

@app.put("/users/{user_id}", response_model=UserResponse)
async def update_user(user_id: int, user_data: UserBase):
    """
    全量更新用户信息
    
    PUT 要求提供完整的资源数据，缺失字段会被设为默认值或 null
    """
    if user_id not in users_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"用户 {user_id} 不存在"
        )
    
    existing = users_db[user_id]
    updated_user = {
        "id": user_id,
        "username": user_data.username,
        "email": user_data.email,
        "full_name": user_data.full_name,
        "is_active": existing["is_active"],
        "created_at": existing["created_at"]
    }
    users_db[user_id] = updated_user
    
    return updated_user


# ==================== PATCH 方法 ====================

@app.patch("/users/{user_id}", response_model=UserResponse)
async def partial_update_user(user_id: int, user_data: UserUpdate):
    """
    部分更新用户信息
    
    PATCH 只更新提供的字段，其他字段保持不变
    """
    if user_id not in users_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"用户 {user_id} 不存在"
        )
    
    existing = users_db[user_id]
    
    # 只更新提供的字段
    if user_data.email is not None:
        existing["email"] = user_data.email
    if user_data.full_name is not None:
        existing["full_name"] = user_data.full_name
    
    return existing


# ==================== DELETE 方法 ====================

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
    """删除用户"""
    if user_id not in users_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"用户 {user_id} 不存在"
        )
    
    del users_db[user_id]
    return None


# ==================== 软删除示例 ====================

@app.delete("/users/{user_id}/soft", response_model=UserResponse)
async def soft_delete_user(user_id: int):
    """软删除用户（推荐）"""
    if user_id not in users_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"用户 {user_id} 不存在"
        )
    
    users_db[user_id]["is_active"] = False
    return users_db[user_id]

2.3 嵌套资源路由

# ==================== 文章与评论的嵌套资源 ====================

class Comment(BaseModel):
    id: int
    content: str
    author_id: int
    post_id: int
    created_at: str


# 获取文章的所有评论
@app.get("/posts/{post_id}/comments", response_model=List[Comment])
async def list_post_comments(
    post_id: int,
    skip: int = 0,
    limit: int = 100
):
    """获取指定文章的所有评论"""
    # 实现逻辑...
    return []


# 在文章下创建评论
@app.post(
    "/posts/{post_id}/comments",
    response_model=Comment,
    status_code=status.HTTP_201_CREATED
)
async def create_post_comment(post_id: int, content: str):
    """在指定文章下创建评论"""
    # 实现逻辑...
    return {"id": 1, "content": content, "post_id": post_id}


# 获取单条评论
@app.get("/posts/{post_id}/comments/{comment_id}", response_model=Comment)
async def get_post_comment(post_id: int, comment_id: int):
    """获取指定文章下的指定评论"""
    # 实现逻辑...
    return {}


# 替代方案：扁平化路由（某些场景更简洁）
@app.get("/comments")
async def list_comments(
    post_id: Optional[int] = None,  # 通过查询参数筛选
    author_id: Optional[int] = None
):
    """获取评论列表，可通过 post_id 或 author_id 筛选"""
    # 实现逻辑...
    return []

3. 状态码与错误处理

3.1 HTTP 状态码规范

2xx 成功
状态码 含义 使用场景
200 OK 请求成功 GET, PUT, PATCH 成功
201 Created 创建成功 POST 创建资源成功
204 No Content 无返回内容 DELETE 成功
3xx 重定向
状态码 含义 使用场景
301 Moved Permanently 永久重定向 资源 URL 变更
304 Not Modified 未修改 缓存有效
4xx 客户端错误
状态码 含义 使用场景
400 Bad Request 请求格式错误 参数验证失败
401 Unauthorized 未认证 缺少身份凭证
403 Forbidden 禁止访问 权限不足
404 Not Found 资源不存在 资源未找到
409 Conflict 资源冲突 重复创建
422 Unprocessable 语义错误 业务逻辑错误
429 Too Many Requests 请求过多 限流触发
5xx 服务器错误
状态码 含义 使用场景
500 Internal Error 服务器内部错误 未预期的异常
502 Bad Gateway 网关错误 上游服务异常
503 Service Unavailable 服务不可用 维护或过载

3.2 统一的错误响应格式

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel
from typing import Any, Optional
import traceback
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# ==================== 错误响应模型 ====================

class ErrorDetail(BaseModel):
    """错误详情"""
    field: Optional[str] = None
    message: str
    code: Optional[str] = None


class ErrorResponse(BaseModel):
    """统一错误响应格式"""
    success: bool = False
    error_code: str
    message: str
    details: Optional[list] = None
    timestamp: str
    path: Optional[str] = None
    request_id: Optional[str] = None


# ==================== 自定义异常 ====================

class APIException(Exception):
    """API 异常基类"""
    def __init__(
        self,
        message: str,
        error_code: str = "INTERNAL_ERROR",
        status_code: int = 500,
        details: Optional[list] = None
    ):
        self.message = message
        self.error_code = error_code
        self.status_code = status_code
        self.details = details or []


class NotFoundException(APIException):
    """资源不存在异常"""
    def __init__(self, resource: str, resource_id: Any):
        super().__init__(
            message=f"{resource} '{resource_id}' 不存在",
            error_code="RESOURCE_NOT_FOUND",
            status_code=404
        )


class ValidationException(APIException):
    """验证异常"""
    def __init__(self, message: str, details: Optional[list] = None):
        super().__init__(
            message=message,
            error_code="VALIDATION_ERROR",
            status_code=400,
            details=details
        )


class ConflictException(APIException):
    """资源冲突异常"""
    def __init__(self, message: str):
        super().__init__(
            message=message,
            error_code="RESOURCE_CONFLICT",
            status_code=409
        )


class UnauthorizedException(APIException):
    """未授权异常"""
    def __init__(self, message: str = "未提供有效的认证信息"):
        super().__init__(
            message=message,
            error_code="UNAUTHORIZED",
            status_code=401
        )


class ForbiddenException(APIException):
    """禁止访问异常"""
    def __init__(self, message: str = "权限不足"):
        super().__init__(
            message=message,
            error_code="FORBIDDEN",
            status_code=403
        )


class RateLimitException(APIException):
    """限流异常"""
    def __init__(self, retry_after: int = 60):
        super().__init__(
            message=f"请求过于频繁，请 {retry_after} 秒后重试",
            error_code="RATE_LIMIT_EXCEEDED",
            status_code=429
        )
        self.retry_after = retry_after


# ==================== 全局异常处理器 ====================

from datetime import datetime
import uuid


def create_error_response(
    request: Request,
    error_code: str,
    message: str,
    status_code: int,
    details: Optional[list] = None
) -> JSONResponse:
    """创建统一错误响应"""
    return JSONResponse(
        status_code=status_code,
        content={
            "success": False,
            "error_code": error_code,
            "message": message,
            "details": details,
            "timestamp": datetime.utcnow().isoformat(),
            "path": str(request.url.path),
            "request_id": str(uuid.uuid4())[:8]
        }
    )


@app.exception_handler(APIException)
async def api_exception_handler(request: Request, exc: APIException):
    """处理自定义 API 异常"""
    return create_error_response(
        request=request,
        error_code=exc.error_code,
        message=exc.message,
        status_code=exc.status_code,
        details=exc.details
    )


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request,
    exc: RequestValidationError
):
    """处理请求验证错误"""
    details = []
    for error in exc.errors():
        details.append({
            "field": ".".join(str(x) for x in error["loc"]),
            "message": error["msg"],
            "code": error.get("type", "validation_error")
        })
    
    return create_error_response(
        request=request,
        error_code="VALIDATION_ERROR",
        message="请求参数验证失败",
        status_code=422,
        details=details
    )


@app.exception_handler(404)
async def not_found_handler(request: Request, exc):
    """处理 404 错误"""
    return create_error_response(
        request=request,
        error_code="ENDPOINT_NOT_FOUND",
        message=f"接口 {request.url.path} 不存在",
        status_code=404
    )


@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """处理所有未捕获的异常"""
    # 记录错误日志
    logger.error(f"未处理的异常: {str(exc)}")
    logger.error(traceback.format_exc())
    
    # 生产环境不暴露详细错误信息
    return create_error_response(
        request=request,
        error_code="INTERNAL_SERVER_ERROR",
        message="服务器内部错误",
        status_code=500
    )


# ==================== 使用示例 ====================

@app.get("/users/{user_id}")
async def get_user_v2(user_id: int, request: Request):
    """使用自定义异常的用户查询"""
    if user_id <= 0:
        raise ValidationException(
            message="用户ID必须大于0",
            details=[{"field": "user_id", "message": "必须是正整数"}]
        )
    
    if user_id not in users_db:
        raise NotFoundException(resource="用户", resource_id=user_id)
    
    return {"success": True, "data": users_db[user_id]}


@app.post("/users")
async def create_user_v2(user_data: UserCreate):
    """使用自定义异常的用户创建"""
    # 检查用户名冲突
    for user in users_db.values():
        if user["username"] == user_data.username:
            raise ConflictException(f"用户名 '{user_data.username}' 已被使用")
    
    # 创建用户逻辑...
    return {"success": True, "data": {"id": 1}}

3.3 成功响应格式

from typing import Generic, TypeVar, Optional
from pydantic.generics import GenericModel

T = TypeVar("T")


class SuccessResponse(GenericModel, Generic[T]):
    """统一成功响应格式"""
    success: bool = True
    data: T
    meta: Optional[dict] = None
    timestamp: str


class PaginatedData(BaseModel, Generic[T]):
    """分页数据结构"""
    items: List[T]
    total: int
    page: int
    page_size: int
    total_pages: int
    has_next: bool
    has_prev: bool


def create_success_response(
    data: Any,
    meta: Optional[dict] = None
) -> dict:
    """创建成功响应"""
    return {
        "success": True,
        "data": data,
        "meta": meta,
        "timestamp": datetime.utcnow().isoformat()
    }


def create_paginated_response(
    items: List[Any],
    total: int,
    page: int,
    page_size: int
) -> dict:
    """创建分页响应"""
    total_pages = (total + page_size - 1) // page_size
    
    return create_success_response(
        data={
            "items": items,
            "total": total,
            "page": page,
            "page_size": page_size,
            "total_pages": total_pages,
            "has_next": page < total_pages,
            "has_prev": page > 1
        }
    )


# 使用示例
@app.get("/users", response_model=SuccessResponse)
async def list_users_v2(
    page: int = 1,
    page_size: int = 20,
    sort_by: str = "created_at",
    sort_order: str = "desc"
):
    """分页查询用户"""
    user_list = list(users_db.values())
    total = len(user_list)
    
    # 分页逻辑
    start = (page - 1) * page_size
    end = start + page_size
    items = user_list[start:end]
    
    return create_paginated_response(
        items=items,
        total=total,
        page=page,
        page_size=page_size
    )

4. API 版本控制策略

4.1 版本控制方式对比

方式 示例 优点 缺点
URL 路径 /v1/users 直观，易于缓存 URL 变长
请求头 Accept: application/vnd.api.v1+json URL 简洁 不够直观
查询参数 /users?version=1 简单 不符合 REST 风格
主机名 api-v1.example.com 可独立部署 配置复杂

4.2 URL 路径版本控制（推荐）

from fastapi import APIRouter

# ==================== v1 版本 ====================

router_v1 = APIRouter(prefix="/v1")


class UserV1(BaseModel):
    id: int
    name: str
    email: str


@router_v1.get("/users", response_model=List[UserV1])
async def list_users_v1():
    """v1 版本：基础用户列表"""
    return [
        {"id": 1, "name": "张三", "email": "zhangsan@example.com"}
    ]


@router_v1.get("/users/{user_id}", response_model=UserV1)
async def get_user_v1(user_id: int):
    """v1 版本：获取用户详情"""
    return {"id": user_id, "name": "张三", "email": "zhangsan@example.com"}


# ==================== v2 版本 ====================

router_v2 = APIRouter(prefix="/v2")


class UserV2(BaseModel):
    id: int
    first_name: str
    last_name: str
    email: str
    phone: Optional[str] = None
    address: Optional[dict] = None
    metadata: Optional[dict] = None


@router_v2.get("/users", response_model=List[UserV2])
async def list_users_v2(
    include_metadata: bool = False,
    fields: Optional[str] = None
):
    """
    v2 版本：增强用户列表
    
    - include_metadata: 是否包含元数据
    - fields: 指定返回的字段（字段过滤）
    """
    users = [
        {
            "id": 1,
            "first_name": "三",
            "last_name": "张",
            "email": "zhangsan@example.com",
            "phone": "13800138000",
            "address": {
                "city": "北京",
                "street": "长安街"
            },
            "metadata": {"vip": True} if include_metadata else None
        }
    ]
    
    # 字段过滤
    if fields:
        field_list = fields.split(",")
        users = [
            {k: v for k, v in user.items() if k in field_list}
            for user in users
        ]
    
    return users


@router_v2.get("/users/{user_id}", response_model=UserV2)
async def get_user_v2(user_id: int):
    """v2 版本：获取更详细的用户信息"""
    return {
        "id": user_id,
        "first_name": "三",
        "last_name": "张",
        "email": "zhangsan@example.com",
        "phone": "13800138000",
        "address": {
            "city": "北京",
            "street": "长安街"
        },
        "metadata": {"vip": True, "level": "gold"}
    }


# ==================== 注册路由 ====================

app = FastAPI(
    title="版本化 API 示例",
    version="2.0.0",
    description=""\
# python # 前端 # restful
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

MCP协议开发实战：从零搭建AI Agent工具链

本文介绍了MCP协议在AI工具链中的应用与实践。MCP协议旨在解决异构工具集成难题，为AI模型提供标准化、安全的工具调用接口。文章详细解析了MCP的核心架构，包括Server、Client、Transport等组件，以及通信流程与安全模型。通过实战示例，指导读者开发计算器和文件系统工具Server，并集成到Claude等AI客户端中。进阶部分展示了如何构建复杂工具链，包括API集成、数据库查询和多

avatar AI Agent技术社区
cover

深入理解 AI Agent · AGENT #02：Agent 的四大核心机制，拆开来看每个都在干什么

avatar AI Agent技术社区
cover

AI Tool 设计：粒度、参数与错误恢复怎么做

avatar AI Agent技术社区