FastAPI: Phase 1 综合练习 — 搭建 PriceTracker 基础 API

Q: 如何测试 OpenAPI 文档是否可被前端消费？

访问 /openapi.json ，用 npx openapi-typescript 生成 TypeScript 类型，或用 Swagger Codegen 生成 SDK。

Q: 多个端点共享模型时如何避免重复？

用模型继承： ProductBase(BaseModel) 定义公共字段， ProductCreate(ProductBase) 和 ProductResponse(ProductBase) 继承并扩展。

最后更新：2026-08-26

Phase 1 的 5 课知识就像五块拼图，现在该把它们拼成一幅完整的画——PriceTracker 基础 API，让 Bob 的前端真正能消费数据。

1. 你将学到

设计并实现 /products、/products/{id}、/prices 三组核心端点
完整的 Pydantic V2 模型体系：ProductCreate/ProductResponse/PriceCreate/PriceResponse
路径参数 + 查询参数 + 请求体的组合实战
使用 response_model 实现"公开查询隐藏批发价"的业务逻辑
Bob 前端对接：确认 OpenAPI 文档可被前端自动生成 SDK 消费

2. Alice 的真实故事

(1) 痛点：零散知识无法组装成产品

Alice 学完了前 5 课，但各课示例都是独立片段——路径参数的例子和 Pydantic 的例子没有串联。Bob 等不及了："我需要一组能用的 API，不是零散的 demo！"Alice 需要把路由设计、参数校验、数据验证、响应过滤整合为一个可运行的 API 服务。

(2) 综合实战的解法

本课将前 5 课所有知识点整合为 PriceTracker 基础 API，包含完整的 CRUD 端点、Pydantic 模型体系、参数校验链和响应过滤逻辑。

PYTHON

# Complete PriceTracker basic API structure
from fastapi import FastAPI, Path, Query, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="PriceTracker API", version="0.1.0")

# Models, routes, validation — all integrated

(3) 收益

Alice 拥有一个可运行的 API 服务，Bob 可以用 Swagger UI 测试所有端点，自动生成的 OpenAPI 文档让前端 SDK 自动生成工具直接消费。

3. API 端点全景设计

(1) Phase 1 端点规划

flowchart LR
    Client[Client] -->|GET /products| List[List Products]
    Client -->|POST /products| Create[Create Product]
    Client -->|GET /products/id| Detail[Product Detail]
    Client -->|PUT /products/id| Update[Update Product]
    Client -->|DELETE /products/id| Delete[Delete Product]
    Client -->|GET /prices| Search[Search Prices]
    Client -->|POST /prices| AddPrice[Add Price]
    
    List --> PM[ProductResponse]
    Create --> PM
    Detail --> PDP[ProductDetailResponse]
    Search --> PRM[PriceResponse]
    AddPrice --> PRM

端点	方法	路径参数	查询参数	请求体	响应模型
商品列表	GET	-	category, sort, limit, offset	-	`list[ProductResponse]`
创建商品	POST	-	-	`ProductCreate`	`ProductResponse`
商品详情	GET	product_id	-	-	`ProductDetailResponse`
更新商品	PUT	product_id	-	`ProductUpdate`	`ProductResponse`
删除商品	DELETE	product_id	-	-	`dict`
价格查询	GET	-	product_id, min_price, max_price	-	`list[PriceResponse]`
添加价格	POST	-	-	`PriceCreate`	`PriceResponse`

▶ 示例：完整 Pydantic 模型体系

PYTHON

from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Optional
from enum import Enum
from datetime import datetime

class Category(str, Enum):
    electronics = "electronics"
    clothing = "clothing"
    food = "food"
    books = "books"

class PriceInfo(BaseModel):
    amount: float = Field(gt=0, description="Price amount in USD")
    currency: str = Field(default="USD", pattern=r"^[A-Z]{3}$")

class ProductCreate(BaseModel):
    name: str = Field(min_length=1, max_length=200)
    category: Category
    base_price: float = Field(gt=0, description="Base price in USD")
    description: Optional[str] = Field(None, max_length=2000)

class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=200)
    category: Optional[Category] = None
    base_price: Optional[float] = Field(None, gt=0)
    description: Optional[str] = Field(None, max_length=2000)

class ProductResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    category: str
    base_price: float

class ProductDetailResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    category: str
    base_price: float
    description: Optional[str] = None
    created_at: datetime

class PriceCreate(BaseModel):
    product_id: int = Field(gt=0)
    price: float = Field(gt=0, description="Price in USD")
    currency: str = Field(default="USD", pattern=r"^[A-Z]{3}$")
    source: str = Field(max_length=100)

    @field_validator("price")
    @classmethod
    def round_price(cls, v: float) -> float:
        return round(v, 2)

class PriceResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    product_id: int
    price: float
    currency: str
    source: str
    recorded_at: datetime

输出：

TEXT 📖 仅展示

# 函数定义成功

4. 请求-响应生命周期

(1) 完整流转过程

sequenceDiagram
    participant Client as Bob Frontend
    participant FastAPI as FastAPI Router
    participant Pydantic as Pydantic Validator
    participant Handler as View Function
    participant DB as In-Memory DB

    Client->>FastAPI: POST /products (JSON body)
    FastAPI->>Pydantic: Validate with ProductCreate
    Pydantic-->>FastAPI: Validated model instance
    FastAPI->>Handler: create_product(data: ProductCreate)
    Handler->>DB: Store product
    DB-->>Handler: Stored record
    Handler-->>FastAPI: Return full dict
    FastAPI->>Pydantic: Filter with ProductResponse
    Pydantic-->>Client: JSON response (filtered)

▶ 示例：商品 CRUD 端点实现

PYTHON

from fastapi import FastAPI, Path, Query, HTTPException
from datetime import datetime

app = FastAPI(title="PriceTracker API", version="0.1.0")

# In-memory storage (will be replaced with DB in Phase 2)
products_db: dict[int, dict] = {}
prices_db: list[dict] = []
_product_counter = 0
_price_counter = 0

@app.post("/products", response_model=ProductResponse, status_code=201)
async def create_product(product: ProductCreate):
    global _product_counter
    _product_counter += 1
    record = {
        "id": _product_counter,
        "name": product.name,
        "category": product.category.value,
        "base_price": product.base_price,
        "description": product.description,
        "created_at": datetime.utcnow(),
    }
    products_db[_product_counter] = record
    return record

@app.get("/products", response_model=list[ProductResponse])
async def list_products(
    category: Optional[Category] = Query(None),
    sort: str = Query("name", pattern="^(name|base_price)$"),
    limit: int = Query(20, ge=1, le=100),
    offset: int = Query(0, ge=0),
):
    items = list(products_db.values())
    if category:
        items = [p for p in items if p["category"] == category.value]
    items.sort(key=lambda p: p.get(sort, ""))
    return items[offset : offset + limit]

@app.get("/products/{product_id}", response_model=ProductDetailResponse)
async def get_product(
    product_id: int = Path(gt=0, description="Product ID"),
):
    if product_id not in products_db:
        raise HTTPException(status_code=404, detail="Product not found")
    return products_db[product_id]

输出：

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：更新与删除端点

PYTHON

@app.put("/products/{product_id}", response_model=ProductResponse)
async def update_product(
    product_id: int = Path(gt=0),
    update: ProductUpdate = ...,
):
    if product_id not in products_db:
        raise HTTPException(status_code=404, detail="Product not found")
    record = products_db[product_id]
    update_data = update.model_dump(exclude_unset=True)
    record.update(update_data)
    return record

@app.delete("/products/{product_id}")
async def delete_product(product_id: int = Path(gt=0)):
    if product_id not in products_db:
        raise HTTPException(status_code=404, detail="Product not found")
    del products_db[product_id]
    return {"message": "Product deleted"}

输出：

TEXT 📖 仅展示

# 函数定义成功

5. 价格端点与组合实战

(1) 查询参数 + 请求体组合

▶ 示例：价格查询与创建端点

PYTHON

@app.get("/prices", response_model=list[PriceResponse])
async def search_prices(
    product_id: Optional[int] = Query(None, gt=0),
    min_price: float = Query(0, ge=0, description="Min price in USD"),
    max_price: float = Query(999999, ge=0, description="Max price in USD"),
    limit: int = Query(50, ge=1, le=200),
    offset: int = Query(0, ge=0),
):
    results = prices_db
    if product_id:
        results = [p for p in results if p["product_id"] == product_id]
    results = [p for p in results if min_price <= p["price"] <= max_price]
    return results[offset : offset + limit]

@app.post("/prices", response_model=PriceResponse, status_code=201)
async def create_price(price: PriceCreate):
    global _price_counter
    if price.product_id not in products_db:
        raise HTTPException(status_code=404, detail="Product not found")
    _price_counter += 1
    record = {
        "id": _price_counter,
        "product_id": price.product_id,
        "price": price.price,
        "currency": price.currency,
        "source": price.source,
        "recorded_at": datetime.utcnow(),
    }
    prices_db.append(record)
    return record

输出：

TEXT 📖 仅展示

# 函数定义成功

6. 响应过滤实战：公开版 vs 管理版

(1) 业务场景：隐藏批发价

PriceTracker 的零售用户只能看到零售价，管理员可以看到批发价和成本价。

▶ 示例：多角色响应模型

PYTHON

class ProductPublicResponse(BaseModel):
    """Public view - hide wholesale and cost prices"""
    id: int
    name: str
    category: str
    retail_price: float

class ProductAdminResponse(BaseModel):
    """Admin view - show full price chain"""
    id: int
    name: str
    category: str
    retail_price: float
    wholesale_price: float
    cost_price: float
    margin_pct: float  # Profit margin percentage

# Extended in-memory storage with pricing tiers
admin_products_db: dict[int, dict] = {}

@app.get("/products/{product_id}/public", response_model=ProductPublicResponse)
async def get_product_public(product_id: int = Path(gt=0)):
    if product_id not in admin_products_db:
        raise HTTPException(status_code=404, detail="Product not found")
    return admin_products_db[product_id]

@app.get("/products/{product_id}/admin", response_model=ProductAdminResponse)
async def get_product_admin(product_id: int = Path(gt=0)):
    if product_id not in admin_products_db:
        raise HTTPException(status_code=404, detail="Product not found")
    return admin_products_db[product_id]

输出：

TEXT 📖 仅展示

# 函数定义成功

❓ 常见问题

Q 内存存储重启后数据丢失怎么办？

A Phase 1 用内存存储简化学习，Phase 2 会替换为 PostgreSQL + SQLAlchemy。这是渐进式学习的策略。

Q ProductCreate 和 ProductResponse 为什么要分开？

A 分离关注点：Create 模型不含 id（服务端生成），Response 模型含 id。这也是安全最佳实践——客户端不应指定 id。

Q exclude_unset=True 在 PUT 端点有什么用？

A 让客户端只更新想改的字段，未提供的字段不被覆盖为 None。结合 ProductUpdate 的 Optional 字段实现部分更新。

Q 如何测试 OpenAPI 文档是否可被前端消费？

A 访问 /openapi.json，用 npx openapi-typescript 生成 TypeScript 类型，或用 Swagger Codegen 生成 SDK。

Q 枚举参数的值怎么在文档中展示？

A FastAPI 自动将 Enum 的所有值展示在 Swagger UI 的下拉选择中，Bob 前端开发者一目了然。

Q 多个端点共享模型时如何避免重复？

A 用模型继承：ProductBase(BaseModel) 定义公共字段，ProductCreate(ProductBase) 和 ProductResponse(ProductBase) 继承并扩展。

📖 小节

Phase 1 完成 PriceTracker 三组核心端点：商品 CRUD、价格查询/创建、公开/管理视图
Pydantic 模型体系：Create（输入校验）、Update（部分更新）、Response（输出过滤）三层分离
路径参数 + 查询参数 + 请求体自然组合，FastAPI 按类型自动区分
response_model 实现"同一数据、不同视图"——公开版隐藏批发价，管理版展示完整价格链
OpenAPI 文档自动生成，Bob 前端可直接用 Swagger UI 测试或生成 SDK

📝 作业

基础题（难度⭐）：将本课所有代码整合到一个 app/main.py 中，启动服务，用 Swagger UI 创建 3 个商品并查询列表。提示：uvicorn app.main:app --reload
进阶题（难度⭐⭐）：添加 /products/{id}/prices 端点，查询指定商品的所有价格记录，支持 sort（date/price）和 limit 查询参数。提示：product_id 路径参数 + sort/limit 查询参数
挑战题（难度⭐⭐⭐）：实现 ProductAdmin 和 ProductPublic 双视图系统——创建含完整价格链的商品数据，公开端点只返回零售价，管理端点返回零售价+批发价+利润率。提示：两个 response_model + 同一数据源

---|

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

关于我们

FastAPI: Phase 1 综合练习 — 搭建 PriceTracker 基础 API

1. 你将学到

2. Alice 的真实故事

(1) 痛点：零散知识无法组装成产品

(2) 综合实战的解法

(3) 收益

3. API 端点全景设计

(1) Phase 1 端点规划

▶ 示例：完整 Pydantic 模型体系

4. 请求-响应生命周期

(1) 完整流转过程

▶ 示例：商品 CRUD 端点实现

▶ 示例：更新与删除端点

5. 价格端点与组合实战

(1) 查询参数 + 请求体组合

▶ 示例：价格查询与创建端点

6. 响应过滤实战：公开版 vs 管理版

(1) 业务场景：隐藏批发价

▶ 示例：多角色响应模型

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好