FastAPI: 项目设计 — PriceTracker 全栈架构蓝图

Q: DDD 的聚合根有什么用？

聚合根定义事务边界和一致性保证。Product 是聚合根意味着 Price 只能通过 Product 修改，不能独立增删，确保数据一致性。

Q: PostgreSQL 读写分离怎么做？

主库写、从库读。SQLAlchemy 配置两个 engine（write_engine + read_engine），读操作用 read_engine 连从库。需处理复制延迟。

Q: 为什么选 Redis 不选 Memcached？

Redis 支持持久化、多种数据结构（Hash/Set/ZSet）、Pub/Sub，Memcached 只支持简单 KV。Redis 功能覆盖更广。

Q: API 版本何时升级？

只在破坏性变更时升版本（如删除字段、改变响应结构）。新增字段和端点不算破坏性变更，不需要新版本。

Q: 领域事件怎么实现？

简单方案用 Celery 任务（如 publish_price_updated_event.delay(product_id, new_price) ），复杂方案用消息队列（RabbitMQ/Kafka）。

Q: 架构图应该多详细？

对团队来说"够用"就行。过度设计浪费时间，设计不足导致混乱。PriceTracker 用三层架构+聚合根划分恰到好处。

最后更新：2026-08-26

项目设计就像盖楼前的蓝图——没有蓝图的施工，盖到三层发现地基不够，只能推倒重来。好蓝图不是画完就扔的，而是施工全程的导航。

1. 你将学到

需求分析：SaaS 价格追踪的核心用例（用户、商品、价格、订阅、通知）
领域驱动设计：聚合根、值对象、领域事件的识别
技术选型决策：FastAPI + PostgreSQL + Redis + Celery + WebSocket 的选型依据
API 版本化策略：URL 版本 vs Header 版本的权衡
架构蓝图：Charlie 的部署拓扑——负载均衡、数据库读写分离、缓存分层

2. Alice 的真实故事

(1) 痛点：功能越加越乱

PriceTracker 从一个简单 API 演变成功能越来越多的服务，但没有统一设计——商品和价格的边界模糊，订阅逻辑散落在 10 个端点，通知系统与业务逻辑耦合。每次加功能都要改 5 个文件，回归 Bug 频发。

(2) 领域驱动设计的解法

领域驱动设计（DDD）从业务视角识别核心领域和边界：商品聚合、价格聚合、用户聚合、订阅聚合——每个聚合有清晰的边界和职责，跨聚合通过事件通信，不再互相纠缠。

(3) 收益

新功能只需修改对应聚合内的代码，不再牵一发动全身。订阅逻辑集中在 Subscription 聚合，通知通过 PriceUpdated 事件触发，解耦彻底。

3. 需求分析

(1) 核心用例

角色	用例	优先级
用户（Alice）	注册/登录	P0
用户	查询商品价格	P0
用户	订阅价格变动通知	P1
管理员	批量导入价格	P0
管理员	管理商品 CRUD	P0
前端（Bob）	调用 API 获取数据	P0
DevOps（Charlie）	监控系统健康	P1
系统	自动爬取价格	P2

▶ 示例：用例优先级排序脚本

PYTHON

# scripts/prioritize_use_cases.py
use_cases = [
    {"role": "User", "action": "register_login", "priority": "P0", "effort": 2},
    {"role": "User", "action": "query_price", "priority": "P0", "effort": 1},
    {"role": "User", "action": "subscribe_alert", "priority": "P1", "effort": 3},
    {"role": "Admin", "action": "bulk_import", "priority": "P0", "effort": 5},
]

# 按优先级排序，同优先级按工作量升序
order = {"P0": 0, "P1": 1, "P2": 2}
sorted_cases = sorted(use_cases, key=lambda x: (order[x["priority"]], x["effort"]))
for uc in sorted_cases:
    print(f"[{uc['priority']}] {uc['role']}: {uc['action']} (effort={uc['effort']}d)")

输出：

TEXT 📖 仅展示

# 执行成功

(2) 非功能性需求

需求	目标	约束
QPS	百万级（集群）	Nginx LB + K8s 弹性
P99 延迟	< 50ms	Redis 缓存 + 异步 DB
可用性	99.9%	多副本 + 自动恢复
数据量	百万商品 + 千万价格	PostgreSQL 分区

4. 领域模型设计

(1) ER 图

erDiagram
    User ||--o{ Subscription : has
    User ||--o{ Product : creates
    User ||--o{ Alert : sets
    Product ||--o{ Price : has
    Product ||--o{ Alert : watched_by
    Subscription ||--o{ Feature : includes
    
    User {
        int id PK
        string email UK
        string hashed_password
        string role
        datetime created_at
    }
    
    Product {
        int id PK
        string name
        string category
        float base_price
        string sku UK
        int user_id FK
        datetime created_at
    }
    
    Price {
        int id PK
        int product_id FK
        float price
        string currency
        string source
        datetime recorded_at
    }
    
    Subscription {
        int id PK
        int user_id FK
        string plan
        datetime starts_at
        datetime expires_at
    }
    
    Alert {
        int id PK
        int user_id FK
        int product_id FK
        float target_price
        string status
    }

(2) 聚合根识别

聚合根	包含实体	边界规则
User	User, Subscription, Alert	用户 owns 订阅和告警
Product	Product, Price	商品 owns 价格历史
PriceImport	批次信息, 导入状态	批量导入是独立事务

▶ 示例：领域事件定义

PYTHON

# domain/events.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class EventType(Enum):
    PRICE_CHANGED = "price_changed"
    ALERT_TRIGGERED = "alert_triggered"
    USER_SUBSCRIBED = "user_subscribed"

@dataclass
class DomainEvent:
    event_type: EventType
    aggregate_id: int
    occurred_at: datetime = field(default_factory=datetime.utcnow)
    payload: dict = field(default_factory=dict)

# 用法：价格变动事件
price_event = DomainEvent(
    event_type=EventType.PRICE_CHANGED,
    aggregate_id=42,
    payload={"old_price": 299.9, "new_price": 249.9, "currency": "CNY"},
)

输出：

TEXT 📖 仅展示

# 执行成功

5. 技术选型决策

(1) 选型对比

需求	选项 A	选项 B	决策	理由
Web 框架	FastAPI	Flask/Django	FastAPI	异步+自动文档+类型校验
数据库	PostgreSQL	MySQL/MongoDB	PostgreSQL	JSONB+异步驱动成熟
缓存	Redis	Memcached	Redis	数据结构丰富+持久化
任务队列	Celery+Redis	RQ/Dramatiq	Celery	生态成熟+监控完善
实时通信	WebSocket	SSE	WebSocket	双向+低延迟
容器	Docker	裸机/VM	Docker	一致性+编排

(2) 系统架构全景

flowchart TD
    Client[Client / Bob Frontend] --> Nginx[Nginx Load Balancer]
    Nginx --> API1[FastAPI Pod 1]
    Nginx --> API2[FastAPI Pod 2]
    Nginx --> APIN[FastAPI Pod N]
    
    API1 --> PG_Master[(PostgreSQL Master)]
    API2 --> PG_Master
    APIN --> PG_Master
    
    PG_Master --> PG_Replica[(PostgreSQL Replica)]
    
    API1 --> Redis[(Redis Cluster)]
    API2 --> Redis
    APIN --> Redis
    
    Redis --> CW1[Celery Worker 1]
    Redis --> CW2[Celery Worker 2]
    
    CW1 --> PG_Master
    CW2 --> PG_Master
    
    API1 --> WS[WebSocket Hub]
    API2 --> WS
    
    Prometheus[Prometheus] --> API1
    Grafana[Grafana] --> Prometheus

6. API 版本化策略

(1) URL 版本 vs Header 版本

维度	URL 版本 `/api/v1/`	Header 版本 `Accept: v=1`
可见性	高（URL 明确）	低（隐藏在 Header）
路由	简单（前缀区分）	复杂（需中间件解析）
缓存	URL 独立缓存	需 Vary Header
Swagger	自动分组	需手动配置
推荐	✅ PriceTracker 使用	适合内部 API

▶ 示例：URL 版本化路由结构

PYTHON

from fastapi import APIRouter

# V1 routes
v1_router = APIRouter(prefix="/api/v1", tags=["v1"])
v1_products = APIRouter(prefix="/products", tags=["products"])
v1_prices = APIRouter(prefix="/prices", tags=["prices"])
v1_auth = APIRouter(prefix="/auth", tags=["auth"])

# V2 routes (future)
v2_router = APIRouter(prefix="/api/v2", tags=["v2"])

# Mount routers
app.include_router(v1_auth)
app.include_router(v1_products, dependencies=[Depends(get_current_user)])
app.include_router(v1_prices, dependencies=[Depends(get_current_user)])
app.include_router(v1_router)

输出：

TEXT 📖 仅展示

# 执行成功

❓ 常见问题

Q DDD 的聚合根有什么用？

A 聚合根定义事务边界和一致性保证。Product 是聚合根意味着 Price 只能通过 Product 修改，不能独立增删，确保数据一致性。

Q PostgreSQL 读写分离怎么做？

A 主库写、从库读。SQLAlchemy 配置两个 engine（write_engine + read_engine），读操作用 read_engine 连从库。需处理复制延迟。

Q 为什么选 Redis 不选 Memcached？

A Redis 支持持久化、多种数据结构（Hash/Set/ZSet）、Pub/Sub，Memcached 只支持简单 KV。Redis 功能覆盖更广。

Q API 版本何时升级？

A 只在破坏性变更时升版本（如删除字段、改变响应结构）。新增字段和端点不算破坏性变更，不需要新版本。

Q 领域事件怎么实现？

A 简单方案用 Celery 任务（如 publish_price_updated_event.delay(product_id, new_price)），复杂方案用消息队列（RabbitMQ/Kafka）。

Q 架构图应该多详细？

A 对团队来说"够用"就行。过度设计浪费时间，设计不足导致混乱。PriceTracker 用三层架构+聚合根划分恰到好处。

📖 小节

需求分析区分功能性和非功能性需求，明确优先级和约束
DDD 识别聚合根（User、Product、PriceImport），定义事务边界和一致性规则
技术选型基于场景决策：FastAPI（异步）+ PostgreSQL（JSONB）+ Redis（数据结构）+ Celery（任务队列）
API 版本化用 URL 前缀 /api/v1/，简单直观、缓存友好
部署拓扑：Nginx LB → FastAPI Pods → PostgreSQL 主从 → Redis Cluster → Celery Workers

📝 作业

基础题（难度⭐）：画出 PriceTracker 的系统架构图（含 FastAPI、PostgreSQL、Redis、Celery、Nginx），标注每个组件的职责和数据流向。提示：Mermaid flowchart
进阶题（难度⭐⭐）：设计 PriceTracker 的领域模型 ER 图，识别 3 个聚合根（User、Product、PriceImport），定义每个聚合包含的实体和边界规则。提示：Mermaid erDiagram + 聚合根表
挑战题（难度⭐⭐⭐）：完成 PriceTracker 完整架构蓝图——需求文档（核心用例+非功能性需求）、技术选型对比表、API 版本化路由结构代码、部署拓扑图（含主从数据库、缓存分层、Celery 集群）。提示：综合本课所有内容

---|

Web-Tutorial 技术团队

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

关于我们

FastAPI: 项目设计 — PriceTracker 全栈架构蓝图

1. 你将学到

2. Alice 的真实故事

(1) 痛点：功能越加越乱

(2) 领域驱动设计的解法

(3) 收益

3. 需求分析

(1) 核心用例

▶ 示例：用例优先级排序脚本

(2) 非功能性需求

4. 领域模型设计

(1) ER 图

(2) 聚合根识别

▶ 示例：领域事件定义

5. 技术选型决策

(1) 选型对比

(2) 系统架构全景

6. API 版本化策略

(1) URL 版本 vs Header 版本

▶ 示例：URL 版本化路由结构

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好