FastAPI 系列 · 第 2 篇：路由与请求模型——用 Pydantic 重新定义接口契约

适合人群：熟悉 Java Spring Boot 的后端工程师，已完成第 01 篇的环境搭建
阅读时间：约 25 分钟
一句话定位：FastAPI 的路由系统以 Python 类型注解为核心，配合 Pydantic v2 的 Schema 声明，实现了比 Spring MVC 注解体系更简洁却同样严谨的接口契约机制；本篇为 shop-api 搭建完整的商品 CRUD 路由骨架。

---

一、路径参数与查询参数

在 Spring Boot 中，@PathVariable 和 @RequestParam 是声明路由参数的两把钥匙。FastAPI 将这套逻辑统一到函数签名本身——参数名与 URL 模板匹配，则为路径参数；否则为查询参数。框架在运行时自动完成类型转换与校验，无需额外注解。

1.1 路径参数（Path Parameter）

# routers/products.py
from fastapi import APIRouter, Path, Query, HTTPException, status

router = APIRouter(prefix="/products", tags=["商品"])

# 类比 Spring Boot: @GetMapping("/{productId}")
#                   public Product getProduct(@PathVariable Long productId)
@router.get("/{product_id}", response_model=ProductResponse)
async def get_product(
    product_id: int = Path(
        ...,                        # ... 表示必填（Spring Boot 中 required=true 是默认行为）
        ge=1,                       # greater than or equal，类比 @Min(1)
        description="商品 ID",
    )
):
    ...

Path(...) 是 FastAPI 提供的声明式约束对象，底层由 Pydantic v2 驱动。ge=1 等价于 @Min(1) Bean Validation 注解。常用约束汇总如下：

FastAPI 约束参数	等价 Bean Validation	说明
ge=1	@Min(1)	大于等于 1
gt=0	@Positive	严格大于 0
le=100	@Max(100)	小于等于 100
lt=100	@DecimalMax("100", inclusive=false)	严格小于 100
min_length=1	@NotBlank	字符串最短长度
max_length=100	@Size(max=100)	字符串最长长度
pattern=r"^\d{6}$"	@Pattern(regexp="^\\d{6}$")	正则校验（Pydantic v2 用 pattern，v1 用 regex）

1.2 查询参数（Query Parameter）

# 类比 Spring Boot:
# @GetMapping
# public Page<Product> listProducts(
#     @RequestParam(defaultValue = "1") int page,
#     @RequestParam(defaultValue = "20") int size,
#     @RequestParam(required = false) String category,
#     @RequestParam(required = false) String keyword
# )
@router.get("", response_model=ProductListResponse)
async def list_products(
    page: int = Query(1, ge=1, description="页码，从 1 开始"),
    size: int = Query(20, ge=1, le=100, description="每页数量，最大 100"),
    category: str | None = Query(None, max_length=50, description="按分类过滤"),
    keyword: str | None = Query(None, max_length=50, description="商品名称关键词"),
):
    # category 和 keyword 类型为 str | None，等价于 @RequestParam(required=false)
    ...

💡 参数有无默认值决定是否必填：Query(...) 表示必填，Query(None) 或 Query(1) 表示可选并携带默认值，与 Spring Boot 的 required 属性语义一致。

1.3 路径与查询参数的类型自动转换

FastAPI 会按照函数签名中声明的 Python 类型，对原始字符串做自动转换：

# 请求：GET /products/42?page=2&active=true
@router.get("/{product_id}")
async def demo(
    product_id: int,      # "42" -> 42
    page: int = 1,        # "2"  -> 2
    active: bool = True,  # "true" / "1" / "yes" -> True（FastAPI 做了宽松转换）
):
    print(type(product_id))  # <class 'int'>

⚠️ 注意：若类型转换失败（如 /products/abc），FastAPI 会自动返回 422 Unprocessable Entity，响应体中包含字段级错误详情，无需开发者手动处理，类比 Spring Boot 的 MethodArgumentTypeMismatchException 全局处理。

---

二、请求体与 Pydantic v2

Spring Boot 通过 @RequestBody + Jackson 反序列化请求体，并结合 Bean Validation (@Valid / @Validated) 做字段校验。FastAPI 将这两件事合并为一步——用 Pydantic Model 声明 Schema，框架负责反序列化与校验，且校验逻辑用 Python 代码而非注解表达，更易测试与复用。

以下是 shop-api 商品相关 Schema 的继承关系图：

2.1 BaseModel 基础

Pydantic 是 Python 生态最流行的数据校验与序列化库，FastAPI 在其基础上构建了整套请求/响应处理管道。v2 版本底层用 Rust（pydantic-core）重写，性能相比 v1 提升 5-50 倍。

# schemas/product.py
from datetime import datetime
from decimal import Decimal
from typing import Optional
from pydantic import BaseModel, Field, model_validator, ConfigDict

# ProductBase 是共享字段的抽象基类，类比 Java 中提取的 DTO 抽象父类
class ProductBase(BaseModel):
    name: str = Field(
        ...,                        # 必填字段，类比 @NotBlank
        min_length=1,
        max_length=100,
        description="商品名称",
    )
    price: Decimal = Field(
        ...,
        gt=0,                       # 类比 @Positive
        decimal_places=2,           # 保留两位小数
        description="商品售价（元）",
    )
    stock: int = Field(
        0,                          # 有默认值，类比 @Column(columnDefinition="INT DEFAULT 0")
        ge=0,
        description="库存数量",
    )
    category: str = Field(
        ...,
        max_length=50,
        description="商品分类",
    )
    description: Optional[str] = Field(
        None,                       # Optional 字段，类比 @RequestParam(required=false)
        max_length=1000,
        description="商品描述",
    )

2.2 继承与扩展——创建、更新、响应三态分离

在 Spring Boot 中，同一个 Product 实体往往对应 CreateDTO、UpdateDTO、ResponseVO 三个不同对象，以避免字段污染。FastAPI/Pydantic 同样推荐这一模式：

# --- 创建请求体 ---
# 类比 Spring Boot 的 ProductCreateDTO + @RequestBody
class ProductCreate(ProductBase):
    cost_price: Decimal = Field(
        ...,
        gt=0,
        description="成本价（仅内部使用，不对外暴露）",
    )

    @model_validator(mode="after")  # mode="after" 表示在所有字段验证完成后执行
    def validate_price_margin(self) -> "ProductCreate":
        """
        跨字段联合校验：售价必须高于成本价
        类比 Spring Boot 的 @AssertTrue 或自定义 ConstraintValidator
        """
        if self.price <= self.cost_price:
            raise ValueError(
                f"售价（{self.price}）必须高于成本价（{self.cost_price}）"
            )
        return self


# --- 更新请求体（PATCH 语义：所有字段均为可选）---
# 类比 Spring Boot 的 ProductUpdateDTO（字段加 @Nullable）
class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    price: Optional[Decimal] = Field(None, gt=0)
    stock: Optional[int] = Field(None, ge=0)
    category: Optional[str] = Field(None, max_length=50)
    description: Optional[str] = Field(None, max_length=1000)


# --- 响应模型 ---
# 类比 Spring Boot 的 ProductResponseVO + @JsonIgnore（隐藏 cost_price）
class ProductResponse(ProductBase):
    # from_attributes=True：支持从 ORM 对象（SQLAlchemy model）直接转换
    # 类比 Jackson 的 @JsonIgnoreProperties(ignoreUnknown = true)
    model_config = ConfigDict(from_attributes=True)

    id: int
    created_at: datetime
    updated_at: datetime
    # cost_price 不出现在响应模型中，序列化时自动忽略


# --- 列表响应（含分页信息）---
class ProductListResponse(BaseModel):
    items: list[ProductResponse]
    total: int = Field(description="总记录数")
    page: int = Field(description="当前页码")
    size: int = Field(description="每页数量")
    pages: int = Field(description="总页数")

2.3 model_validator 深入解析

model_validator 是 Pydantic v2 提供的跨字段联合校验机制，类比 Spring Boot 中自定义 @Constraint 注解的实现类。它有两种执行模式：

from pydantic import BaseModel, model_validator

class PriceRange(BaseModel):
    min_price: Decimal
    max_price: Decimal

    # mode="before"：在字段类型转换之前执行，接收原始数据字典
    @model_validator(mode="before")
    @classmethod
    def check_raw_data(cls, values: dict) -> dict:
        """
        适用场景：需要在字段解析前预处理数据
        类比 @JsonDeserialize 自定义反序列化器
        """
        # 将字符串格式的价格统一转为数字
        if isinstance(values.get("min_price"), str):
            values["min_price"] = values["min_price"].replace(",", "")
        return values

    # mode="after"：在所有字段验证成功后执行，self 已是合法的 Model 实例
    @model_validator(mode="after")
    def check_price_range(self) -> "PriceRange":
        """
        适用场景：需要跨字段比较或联合约束
        类比 @ScriptAssert 或自定义类级别 ConstraintValidator
        """
        if self.min_price >= self.max_price:
            raise ValueError("最低价必须小于最高价")
        return self

2.4 Field 的 model_config 扩展

class ProductResponse(ProductBase):
    model_config = ConfigDict(
        from_attributes=True,   # 支持 ORM 对象转换（Pydantic v2 替代 v1 的 orm_mode=True）
        json_schema_extra={     # 自定义 OpenAPI Schema 示例（替代 v1 的 schema_extra）
            "example": {
                "id": 1,
                "name": "高性能 SSD 固态硬盘",
                "price": "599.00",
                "stock": 200,
                "category": "存储设备",
                "created_at": "2024-01-15T10:30:00",
            }
        },
    )
    id: int
    created_at: datetime
    updated_at: datetime

---

三、响应模型与序列化控制

FastAPI 的 response_model 参数是控制接口输出字段的核心机制，其作用等价于 Spring Boot 中 @JsonIgnore、@JsonView、Jackson MixIn 的组合——但语义更直观：声明你希望返回什么，多余的字段自动消失。

3.1 response_model 基本用法

# 类比 Spring Boot:
# @GetMapping("/{id}")
# public ProductResponseVO getProduct(@PathVariable Long id) { ... }
# 此处 ProductResponseVO 只含公开字段，cost_price 不在其中
@router.get("/{product_id}", response_model=ProductResponse)
async def get_product(product_id: int):
    # 假设 Service 层返回的数据库对象含 cost_price 字段
    raw_data = {
        "id": product_id,
        "name": "测试商品",
        "price": Decimal("99.00"),
        "cost_price": Decimal("50.00"),  # 成本价
        "stock": 100,
        "category": "测试",
        "created_at": datetime.now(),
        "updated_at": datetime.now(),
    }
    # FastAPI 会根据 response_model=ProductResponse 过滤掉 cost_price
    # 最终响应体中不会包含 cost_price
    return raw_data

3.2 response_model_exclude_unset

response_model_exclude_unset=True 是 PATCH 接口的利器。它的语义是：只序列化客户端显式设置的字段，忽略未传入的 Optional 字段的默认值（None）。

# 场景：PATCH 接口更新部分字段，响应体只返回被修改的字段
@router.patch("/{product_id}", response_model=ProductResponse,
              response_model_exclude_unset=True)
async def patch_product(product_id: int, product: ProductUpdate):
    # 客户端只传了 {"name": "新名称"}
    # 使用 model_dump(exclude_unset=True) 获取实际传入的字段
    update_data = product.model_dump(exclude_unset=True)
    # update_data = {"name": "新名称"}，而非 {"name": "新名称", "price": None, ...}
    # TODO: 第 03 篇接入 Service 层
    ...

3.3 exclude 与 include 精细控制

# 场景：某些场景需要临时排除特定字段
@router.get("/admin/{product_id}", response_model=ProductResponse,
            response_model_exclude={"description"})   # 排除 description 字段
async def get_product_admin(product_id: int):
    ...

# 场景：只返回部分关键字段（类比 SQL SELECT 指定列）
@router.get("/summary", response_model=ProductResponse,
            response_model_include={"id", "name", "price"})   # 只保留这三个字段
async def list_product_summary():
    ...

3.4 响应模型继承体系

# 完整的 Schema 继承关系示意

#                    BaseModel
#                       |
#                  ProductBase           <- 共享字段：name, price, stock, category
#                  /         \
#        ProductCreate    ProductResponse  <- 各自扩展专属字段
#
#        ProductUpdate                    <- 独立 Schema（全部 Optional）

---

四、HTTP 状态码语义

Spring Boot 中通过 ResponseEntity<T> 或 @ResponseStatus 控制 HTTP 状态码。FastAPI 在路由装饰器上用 status_code 参数声明成功响应码，用 HTTPException 表达错误响应——两者分开声明，职责清晰。

4.1 成功状态码声明

from fastapi import status  # 类比 Spring 的 HttpStatus 枚举

# 201 Created：资源创建成功（类比 ResponseEntity.status(201).body(...)）
@router.post("", response_model=ProductResponse,
             status_code=status.HTTP_201_CREATED)
async def create_product(product: ProductCreate):
    ...

# 204 No Content：删除成功，无响应体（类比 ResponseEntity.noContent().build()）
@router.delete("/{product_id}",
               status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(product_id: int):
    ...

fastapi.status 模块与 starlette.status 完全等价，提供全套 HTTP 状态码常量，避免硬编码数字。

4.2 HTTPException 错误处理

from fastapi import HTTPException, status

# 基础用法：404 Not Found
@router.get("/{product_id}", response_model=ProductResponse)
async def get_product(product_id: int):
    product = fake_db.get(product_id)
    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="商品不存在",          # detail 可以是字符串
        )
    return product


# 进阶用法：detail 使用结构化字典（类比 Spring Boot 的 ErrorResponse 对象）
@router.post("", response_model=ProductResponse,
             status_code=status.HTTP_201_CREATED)
async def create_product(product: ProductCreate):
    if await is_name_duplicate(product.name):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail={                        # detail 可以是任意 JSON 可序列化对象
                "code": "PRODUCT_NAME_DUPLICATE",
                "message": f"商品名称 '{product.name}' 已存在",
                "field": "name",
            },
        )
    ...

4.3 自定义异常处理器

对于业务异常，更推荐定义自定义异常类并注册全局处理器，而非在每个路由中 raise HTTPException——这与 Spring Boot 的 @ControllerAdvice 机制完全对应：

# exceptions.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class ProductNotFoundException(Exception):
    def __init__(self, product_id: int):
        self.product_id = product_id

# main.py 中注册（类比 @ControllerAdvice + @ExceptionHandler）
app = FastAPI()

@app.exception_handler(ProductNotFoundException)
async def product_not_found_handler(request: Request, exc: ProductNotFoundException):
    return JSONResponse(
        status_code=404,
        content={
            "code": "PRODUCT_NOT_FOUND",
            "message": f"商品 ID {exc.product_id} 不存在",
        },
    )

---

五、路由分组——APIRouter

当应用规模增长，将所有路由堆在 main.py 中会带来与 Spring Boot 单控制器膨胀同样的维护困境。FastAPI 提供 APIRouter 作为路由分组机制，类比 Spring Boot 的 @Controller + @RequestMapping——每个 Router 是一个独立的路由模块，最终在应用入口统一注册。

5.1 项目结构规划

shop-api/
├── app/
│   ├── main.py               # 应用入口，类比 Spring Boot Application
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── products.py       # 商品路由，类比 ProductController
│   │   ├── orders.py         # 订单路由，类比 OrderController
│   │   └── users.py          # 用户路由，类比 UserController
│   └── schemas/
│       ├── __init__.py
│       └── product.py        # 商品 Schema，类比 ProductDTO/VO
├── pyproject.toml
└── README.md

5.2 APIRouter 声明

# routers/products.py
from fastapi import APIRouter

# prefix：路由前缀，类比 @RequestMapping("/api/v1/products")
# tags：OpenAPI 文档分组标签，类比 @Tag(name="商品管理")
router = APIRouter(
    prefix="/products",
    tags=["商品"],
    responses={                         # 为整个 Router 声明通用错误响应（出现在 Swagger 文档）
        404: {"description": "商品不存在"},
        422: {"description": "请求参数校验失败"},
    },
)

5.3 在 main.py 中注册

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.routers import products, orders, users  # 导入各 Router 模块


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 应用启动时执行（类比 @PostConstruct 或 ApplicationRunner）
    print("shop-api starting up...")
    yield
    # 应用关闭时执行（类比 @PreDestroy）
    print("shop-api shutting down...")


app = FastAPI(
    title="Shop API",
    description="简化版电商 REST API，FastAPI 系列贯穿项目",
    version="0.1.0",
    lifespan=lifespan,
)

# CORS 中间件（类比 Spring Boot 的 CorsConfigurationSource）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由模块（类比 Spring Boot 自动扫描 @Controller）
# prefix="/api/v1" 是统一的 API 版本前缀
app.include_router(products.router, prefix="/api/v1")
app.include_router(orders.router, prefix="/api/v1")
app.include_router(users.router, prefix="/api/v1")

# 最终路由结果：
# GET /api/v1/products
# GET /api/v1/products/{product_id}
# POST /api/v1/products
# ...


@app.get("/health", tags=["系统"])
async def health_check():
    return {"status": "ok", "service": "shop-api"}

5.4 include_router 参数详解

参数	类型	说明	Spring Boot 类比
router	APIRouter	要注册的路由对象	@Controller 类
prefix	str	额外的路径前缀	@RequestMapping
tags	list[str]	覆盖 Router 的 tags	@Tag
dependencies	list	为整个 Router 注入依赖	拦截器 / AOP
responses	dict	覆盖通用响应声明	—
deprecated	bool	标记整个 Router 已废弃	@Deprecated

💡 版本化路由：可通过 prefix 组合实现 API 版本管理：

app.include_router(products_v1.router, prefix="/api/v1")
app.include_router(products_v2.router, prefix="/api/v2")

---

六、贯穿项目实战——商品 CRUD 路由骨架

本节将前述所有知识点整合，为 shop-api 实现完整的商品 CRUD 路由骨架。Service 层调用以 TODO 占位，将在第 03 篇（依赖注入）中正式接入。

6.1 完整 Schema 定义

# app/schemas/product.py
from datetime import datetime
from decimal import Decimal
from typing import Optional
from pydantic import BaseModel, Field, model_validator, ConfigDict


class ProductBase(BaseModel):
    """商品基础 Schema，包含创建和响应共有字段"""

    name: str = Field(
        ...,
        min_length=1,
        max_length=100,
        description="商品名称",
        examples=["高性能 SSD 固态硬盘 512GB"],
    )
    price: Decimal = Field(
        ...,
        gt=0,
        decimal_places=2,
        description="商品售价（元）",
        examples=["599.00"],
    )
    stock: int = Field(
        default=0,
        ge=0,
        description="库存数量",
    )
    category: str = Field(
        ...,
        max_length=50,
        description="商品分类",
        examples=["存储设备"],
    )
    description: Optional[str] = Field(
        default=None,
        max_length=1000,
        description="商品描述",
    )


class ProductCreate(ProductBase):
    """
    商品创建请求体
    类比 Spring Boot 的 ProductCreateRequest（含 @Valid 校验）
    """

    cost_price: Decimal = Field(
        ...,
        gt=0,
        decimal_places=2,
        description="成本价（仅内部使用，不对外暴露）",
    )

    @model_validator(mode="after")
    def validate_price_margin(self) -> "ProductCreate":
        """售价必须高于成本价（跨字段联合校验）"""
        if self.price <= self.cost_price:
            raise ValueError(
                f"售价（{self.price}）必须高于成本价（{self.cost_price}），"
                f"当前价差为 {self.price - self.cost_price}"
            )
        return self


class ProductUpdate(BaseModel):
    """
    商品更新请求体（所有字段均为 Optional，支持 PATCH 语义）
    类比 Spring Boot 的 ProductUpdateRequest（字段均加 @Nullable）
    """

    name: Optional[str] = Field(default=None, min_length=1, max_length=100)
    price: Optional[Decimal] = Field(default=None, gt=0, decimal_places=2)
    stock: Optional[int] = Field(default=None, ge=0)
    category: Optional[str] = Field(default=None, max_length=50)
    description: Optional[str] = Field(default=None, max_length=1000)

    @model_validator(mode="after")
    def validate_at_least_one_field(self) -> "ProductUpdate":
        """至少需要更新一个字段"""
        provided = self.model_dump(exclude_unset=True)
        if not provided:
            raise ValueError("更新请求至少需要包含一个有效字段")
        return self


class ProductResponse(ProductBase):
    """
    商品响应模型
    - cost_price 不出现在此处，自动隐藏（类比 @JsonIgnore）
    - from_attributes=True 支持从 ORM 对象直接转换（类比 @JsonIgnoreProperties）
    """

    model_config = ConfigDict(
        from_attributes=True,
        json_schema_extra={
            "example": {
                "id": 1,
                "name": "高性能 SSD 固态硬盘 512GB",
                "price": "599.00",
                "stock": 200,
                "category": "存储设备",
                "description": "读速 560MB/s，写速 520MB/s",
                "created_at": "2024-01-15T10:30:00",
                "updated_at": "2024-01-15T10:30:00",
            }
        },
    )

    id: int = Field(description="商品唯一 ID")
    created_at: datetime = Field(description="创建时间")
    updated_at: datetime = Field(description="最后更新时间")


class ProductListResponse(BaseModel):
    """商品列表响应（含分页元信息）"""

    items: list[ProductResponse] = Field(description="商品列表")
    total: int = Field(ge=0, description="总记录数")
    page: int = Field(ge=1, description="当前页码")
    size: int = Field(ge=1, description="每页数量")
    pages: int = Field(ge=0, description="总页数")

    @classmethod
    def empty(cls, page: int = 1, size: int = 20) -> "ProductListResponse":
        """创建空列表响应的工厂方法（类比 Page.empty()）"""
        return cls(items=[], total=0, page=page, size=size, pages=0)

6.2 完整路由实现

# app/routers/products.py
"""
商品 CRUD 路由模块
类比 Spring Boot 的 ProductController

当前版本（第 02 篇）：路由骨架，Service 层以 TODO 占位
下一版本（第 03 篇）：接入依赖注入，注入 ProductService
"""
from fastapi import APIRouter, HTTPException, Path, Query, status

from app.schemas.product import (
    ProductCreate,
    ProductListResponse,
    ProductResponse,
    ProductUpdate,
)

# 路由分组声明
# 类比：@RestController + @RequestMapping("/products") + @Tag(name="商品")
router = APIRouter(
    prefix="/products",
    tags=["商品"],
    responses={
        422: {"description": "请求参数或请求体校验失败"},
    },
)


# ============================================================
# GET /products — 商品列表（含分页和过滤）
# 类比：@GetMapping + Page<ProductResponseVO> listProducts(...)
# ============================================================
@router.get(
    "",
    response_model=ProductListResponse,
    summary="获取商品列表",
    description="支持分页、分类过滤和关键词搜索",
)
async def list_products(
    page: int = Query(default=1, ge=1, description="页码，从 1 开始"),
    size: int = Query(default=20, ge=1, le=100, description="每页数量，最大 100"),
    category: str | None = Query(default=None, max_length=50, description="按分类过滤"),
    keyword: str | None = Query(default=None, max_length=50, description="商品名称关键词搜索"),
) -> ProductListResponse:
    # TODO: 第 03 篇接入 ProductService.list_products(page, size, category, keyword)
    return ProductListResponse.empty(page=page, size=size)


# ============================================================
# GET /products/{product_id} — 获取单个商品详情
# 类比：@GetMapping("/{id}") + ProductResponseVO getProduct(@PathVariable Long id)
# ============================================================
@router.get(
    "/{product_id}",
    response_model=ProductResponse,
    summary="获取商品详情",
)
async def get_product(
    product_id: int = Path(
        ...,
        ge=1,
        description="商品 ID，必须为正整数",
    ),
) -> ProductResponse:
    # TODO: 第 03 篇接入 ProductService.get_by_id(product_id)
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail={
            "code": "PRODUCT_NOT_FOUND",
            "message": f"商品 ID {product_id} 不存在",
        },
    )


# ============================================================
# POST /products — 创建商品
# 类比：@PostMapping + @ResponseStatus(HttpStatus.CREATED)
#        ProductResponseVO createProduct(@RequestBody @Valid ProductCreateRequest req)
# ============================================================
@router.post(
    "",
    response_model=ProductResponse,
    status_code=status.HTTP_201_CREATED,
    summary="创建商品",
)
async def create_product(
    product: ProductCreate,              # 类比 @RequestBody + @Valid ProductCreateRequest
) -> ProductResponse:
    # TODO: 第 03 篇接入 ProductService.create(product)
    # 此处 model_validator 已在 Pydantic 层完成：售价 > 成本价
    raise HTTPException(
        status_code=status.HTTP_501_NOT_IMPLEMENTED,
        detail="Service 层将在第 03 篇实现",
    )


# ============================================================
# PUT /products/{product_id} — 更新商品（全量替换）
# 类比：@PutMapping("/{id}")
#        ProductResponseVO updateProduct(@PathVariable Long id,
#                                        @RequestBody @Valid ProductUpdateRequest req)
# ============================================================
@router.put(
    "/{product_id}",
    response_model=ProductResponse,
    summary="更新商品信息",
)
async def update_product(
    product_id: int = Path(..., ge=1, description="商品 ID"),
    product: ProductUpdate,
) -> ProductResponse:
    # TODO: 第 03 篇接入 ProductService.update(product_id, product)
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail={
            "code": "PRODUCT_NOT_FOUND",
            "message": f"商品 ID {product_id} 不存在",
        },
    )


# ============================================================
# DELETE /products/{product_id} — 删除商品
# 类比：@DeleteMapping("/{id}") + @ResponseStatus(HttpStatus.NO_CONTENT)
# ============================================================
@router.delete(
    "/{product_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="删除商品",
    description="删除成功返回 204 No Content，无响应体",
)
async def delete_product(
    product_id: int = Path(..., ge=1, description="商品 ID"),
) -> None:
    # TODO: 第 03 篇接入 ProductService.delete(product_id)
    # 注意：DELETE 成功时不返回任何内容（status_code=204 自动处理）
    ...

6.3 完整目录结构与文件清单

完成本篇后，shop-api 的目录结构如下：

shop-api/
├── app/
│   ├── __init__.py
│   ├── main.py                    # 已完成（第 01 篇），添加了 include_router
│   ├── routers/
│   │   ├── __init__.py
│   │   └── products.py            # 本篇新增，商品 CRUD 路由骨架
│   └── schemas/
│       ├── __init__.py
│       └── product.py             # 本篇新增，商品 Pydantic Schema
├── tests/
│   └── test_products.py           # 下篇补充完整测试
├── pyproject.toml
└── README.md

6.4 启动与验证

# 启动开发服务器
uvicorn app.main:app --reload --port 8000

# 验证路由注册成功
curl http://localhost:8000/api/v1/products
# 响应：{"items":[],"total":0,"page":1,"size":20,"pages":0}

curl http://localhost:8000/api/v1/products/1
# 响应：{"code":"PRODUCT_NOT_FOUND","message":"商品 ID 1 不存在"}

# 查看自动生成的 OpenAPI 文档（类比 springdoc-openapi 的 /swagger-ui.html）
# http://localhost:8000/docs       — Swagger UI
# http://localhost:8000/redoc      — ReDoc

---

七、Pydantic v2 常见坑

7.1 Pydantic v1 → v2 迁移对照表

FastAPI 0.100+ 开始默认使用 Pydantic v2。若从老项目迁移，以下变更是最常见的坑：

功能	Pydantic v1	Pydantic v2	说明
ORM 支持	orm_mode = True	model_config = ConfigDict(from_attributes=True)	Config 类变为 ConfigDict
自定义 Schema 示例	schema_extra = {...}	model_config = ConfigDict(json_schema_extra={...})	同上
字段级校验器	@validator("field")	@field_validator("field")	函数签名变为 (cls, v)
跨字段校验	@root_validator	@model_validator(mode="before"/"after")	mode 参数替代 pre/post
获取字段数据	.dict()	.model_dump()	dict 方法已废弃
JSON 序列化	.json()	.model_dump_json()	json 方法已废弃
构造方法	Product.parse_obj(data)	Product.model_validate(data)	更语义化
更新模型	.copy(update={...})	.model_copy(update={...})	copy 方法已废弃
正则约束	regex=r"..."	pattern=r"..."	Field 参数名变更
字段别名	Field(alias="...")	不变，但 model_config 中可配置全局 alias 生成规则	—

7.2 常见坑与最佳实践

坑 1：Optional 字段在更新场景下的语义混淆

# ❌ 错误：无法区分"客户端没传 price"和"客户端传了 null 想清空 price"
class ProductUpdate(BaseModel):
    price: Optional[Decimal] = None

# 假设客户端发送：{"name": "新名称"}
# price 会被解析为 None，但我们不知道这是"未传"还是"故意置空"

# ✅ 正确：使用 model_dump(exclude_unset=True) 区分"未设置"和"显式 None"
@router.put("/{product_id}", response_model=ProductResponse)
async def update_product(product_id: int, product: ProductUpdate):
    # 只包含客户端实际传入的字段
    update_data = product.model_dump(exclude_unset=True)
    # 客户端发送 {"name": "新名称"}  ->  update_data = {"name": "新名称"}
    # 客户端发送 {"price": null}    ->  update_data = {"price": None}
    ...

坑 2：model_validator 中访问未验证字段

# ❌ 错误：mode='before' 中用 self 访问字段（此时 self 是原始 dict，没有 .price 属性）
@model_validator(mode="before")
@classmethod
def wrong_before_validator(cls, data):
    if data.price <= data.cost_price:  # AttributeError: 'dict' has no attribute 'price'
        raise ValueError("售价必须高于成本价")
    return data

# ✅ 正确写法一：mode='before' 中用 dict 方式访问
@model_validator(mode="before")
@classmethod
def correct_before_validator(cls, data):
    price = data.get("price", 0)
    cost_price = data.get("cost_price", 0)
    if price and cost_price and price <= cost_price:
        raise ValueError("售价必须高于成本价")
    return data

# ✅ 正确写法二：mode='after' 中用 self 访问（推荐，字段已经过各自校验）
@model_validator(mode="after")
def correct_after_validator(self) -> "ProductCreate":
    if self.price <= self.cost_price:
        raise ValueError("售价必须高于成本价")
    return self

坑 3：Decimal 序列化问题

# ❌ 错误：Decimal 默认序列化为字符串，前端收到 "599.00" 而非数字 599.00
class ProductResponse(BaseModel):
    price: Decimal

# ✅ 正确：通过 model_config 的 json_encoders 或直接用 float 类型
# 方案 A：使用 float（精度要求不高时）
class ProductResponse(BaseModel):
    price: float  # 直接序列化为 JSON number

# 方案 B：保留 Decimal 但自定义序列化
from pydantic import field_serializer

class ProductResponse(BaseModel):
    price: Decimal

    @field_serializer("price")
    def serialize_price(self, v: Decimal) -> float:
        return float(v)  # 序列化为 JSON number 类型

坑 4：忘记给 APIRouter 加前缀导致路由冲突

# ❌ 错误：两个 Router 都定义了 GET /，注册顺序决定哪个生效
products_router = APIRouter()
orders_router = APIRouter()

@products_router.get("/")   # 匹配 GET /
async def list_products(): ...

@orders_router.get("/")     # 同样匹配 GET /，被 products 路由遮蔽
async def list_orders(): ...

# ✅ 正确：每个 Router 必须有唯一的 prefix
products_router = APIRouter(prefix="/products")
orders_router = APIRouter(prefix="/orders")

坑 5：status_code=204 时意外返回响应体

# ❌ 错误：204 No Content 不应有响应体，但函数返回了内容
@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(product_id: int):
    return {"message": "删除成功"}  # FastAPI 会忽略此返回值并发送空响应体
    # 虽然不会报错，但会让代码意图不清晰，且某些客户端会报解析错误

# ✅ 正确：204 路由返回 None 或直接 return，并声明返回类型为 None
@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(product_id: int) -> None:
    # TODO: 执行删除操作
    ...  # 不返回任何内容

---

八、进阶：路由依赖与中间件预览

🔬 本节内容将在后续篇章详细展开，此处仅作概念预告。

8.1 路由级依赖（第 03 篇核心）

from fastapi import Depends

# 为整个 Router 注入认证依赖（类比 Spring Security 的 SecurityFilterChain）
router = APIRouter(
    prefix="/products",
    tags=["商品"],
    dependencies=[Depends(verify_token)],  # 所有该 Router 下的路由都需要验证 token
)

8.2 后台任务（第 06 篇）

from fastapi import BackgroundTasks

@router.post("", status_code=status.HTTP_201_CREATED)
async def create_product(product: ProductCreate, background_tasks: BackgroundTasks):
    result = await product_service.create(product)
    # 异步发送通知，不阻塞响应（类比 Spring 的 @Async）
    background_tasks.add_task(notify_inventory_system, result.id)
    return result

---

总结

本篇系统介绍了 FastAPI 路由体系的核心机制。以下表格回顾关键对应关系：

FastAPI 概念	Spring Boot 对应	核心作用
Path(...)	@PathVariable + @Min/@Max	路径参数声明与约束
Query(...)	@RequestParam + @Size/@Pattern	查询参数声明与约束
BaseModel	DTO / Bean Validation	请求/响应数据结构定义
Field(...)	@NotBlank/@Min/@Size 等	字段级约束
model_validator	@AssertTrue / ConstraintValidator	跨字段联合校验
response_model	@JsonIgnore / @JsonView	控制响应输出字段
HTTPException	ResponseStatusException	抛出 HTTP 错误
status_code=201	@ResponseStatus(HttpStatus.CREATED)	声明成功响应码
APIRouter	@Controller + @RequestMapping	路由分组模块化
include_router	Spring MVC 自动扫描 Controller	注册路由模块
ConfigDict(from_attributes=True)	@JsonIgnoreProperties(ignoreUnknown=true)	ORM 对象转换支持
model_dump(exclude_unset=True)	无直接对应（Spring 需手动实现）	PATCH 语义字段过滤

💡 金句：FastAPI 的接口契约不是写在注解里的，而是写在类型里的——当 Python 类型系统足够丰富，注解就成了多余的噪音。

---

参考资料

• FastAPI 官方文档 - Path Parameters
• FastAPI 官方文档 - Query Parameters
• FastAPI 官方文档 - Request Body
• FastAPI 官方文档 - Response Model
• FastAPI 官方文档 - Bigger Applications
• Pydantic v2 官方文档 - Models
• Pydantic v2 官方文档 - Validators
• Pydantic v2 迁移指南
• HTTP 状态码语义 - MDN Web Docs

---

下期预告

第 03 篇：依赖注入——用 FastAPI Depends 构建可测试的 Service 层

Spring Boot 工程师对 @Autowired 和 IoC 容器驾轻就熟。FastAPI 的 Depends 机制以更轻量的方式实现了相同的思想——不需要容器，仅靠函数组合就能完成依赖的声明、注入与替换。

下篇将完成本篇所有 TODO：

• Depends 基础：函数依赖、类依赖、依赖链
• 实现 ProductService，封装业务逻辑
• 用 Depends 注入 Service 到路由函数
• 实现内存存储（为第 04 篇数据库接入做准备）
• 测试中替换依赖（app.dependency_overrides），类比 Spring Boot 的 @MockBean
• 贯穿项目：shop-api 商品 CRUD 接口全面跑通

敬请期待 🎯