FastAPI: Phase 2 综合练习 — PriceTracker 核心功能闭环
最后更新：2026-08-26
Phase 2 就像汽车总装——引擎（数据库）、方向盘（路由）、安全带（认证）、仪表盘（中间件）都已造好，现在组装成一辆能上路的完整汽车。
1. 你将学到
- 完善三层 DI 链：
get_db→get_current_user→require_subscription("pro") - 全端点 JWT 保护：公开端点 vs 认证端点的路由分组策略
- CORS 中间件配置：允许 Bob 的前端域名跨域访问
- 完整 CRUD + 分页 + 关联预加载的端到端测试流程
- Alice 场景验证：Pro 用户可批量导入百万价格数据，Free 用户受限 1000 条
2. Alice 的真实故事
(1) 痛点：功能分散无法联调
Alice 分别实现了中间件、依赖注入、数据库、CRUD、JWT 认证，但每个功能都是独立测试的。当她尝试把所有功能整合到一起时，发现 CORS 中间件和 JWT 认证的执行顺序冲突、DB Session 在 DI 链中被创建了两次、Pro 用户的限流逻辑和权限检查互相覆盖。
(2) 系统化整合的解法
Phase 2 综合练习将所有模块按正确的层级和顺序整合：CORS（最外层）→ 限流 → 认证 → 业务逻辑 → 数据库（最内层），每层职责清晰、互不干扰。
(3) 收益
整合后的 PriceTracker 核心服务可以完整运行：Bob 前端跨域访问成功、所有受保护端点需要 JWT Token、Pro 用户批量导入顺畅、Free 用户被正确限流。
3. Phase 2 架构总览
(1) 完整请求链路
flowchart TD
Client[Bob Frontend] --> CORS[CORS Middleware]
CORS --> RateLimit[Rate Limit Middleware]
RateLimit --> Router[FastAPI Router]
Router --> AuthDI{Auth Required?}
AuthDI -->|Yes| GetDB[get_db Dependency]
AuthDI -->|No| Handler[Handler]
GetDB --> GetUser[get_current_user]
GetUser --> CheckSub[require_subscription]
CheckSub --> Handler
Handler --> Repo[Repository Layer]
Repo --> DB[(PostgreSQL)]
RateLimit -.->|429| Reject[Rate Limited]
GetUser -.->|401| Unauth[Unauthorized]
CheckSub -.->|403| Forbid[Forbidden]
(2) 路由分组策略
| 路由组 | 前缀 | 认证 | 说明 |
|---|---|---|---|
| public | / |
无 | health, docs |
| api_v1 | /api/v1 |
JWT 必选 | 所有业务端点 |
| admin | /api/v1/admin |
JWT + Admin | 管理端点 |
4. 完整代码整合
(1) 应用入口与中间件配置
▶ 示例：main.py 完整配置
PYTHON
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
from app.api.routes import products, prices, auth
from app.core.config import settings
app = FastAPI(
title=settings.app_name,
version="1.0.0",
description="PriceTracker SaaS API - E-commerce price tracking service",
)
# Middleware order: last registered = innermost (executes first on request)
# CORS should be outermost (registered last)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.allowed_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include routers - public first, then protected
app.include_router(auth.router, tags=["auth"])
app.include_router(
products.router,
prefix="/api/v1",
dependencies=[Depends(get_current_user)],
tags=["products"],
)
app.include_router(
prices.router,
prefix="/api/v1",
dependencies=[Depends(get_current_user)],
tags=["prices"],
)
@app.get("/health")
async def health_check():
return {"status": "healthy", "service": "pricetracker"}
输出：
TEXT
📖 仅展示
# 函数定义成功
(2) 三层 DI 链完整实现
▶ 示例：app/core/deps.py
PYTHON
from fastapi import Depends, HTTPException, Header
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from jose import jwt, JWTError
from app.db import async_session
from app.core.config import settings
from app.core.security import verify_password
from app.models import User
from sqlalchemy import select
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
# Level 1: Database session
async def get_db():
async with async_session() as session:
try:
yield session
except Exception:
await session.rollback()
raise
# Level 2: Current user
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db),
):
credentials_exception = HTTPException(
status_code=401,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(
token, settings.secret_key, algorithms=[settings.algorithm]
)
username = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
stmt = select(User).where(User.email == username)
result = await db.execute(stmt)
user = result.scalar_one_or_none()
if user is None:
raise credentials_exception
return user
# Level 3: Subscription check
def require_subscription(min_level: str = "free"):
LEVELS = {"free": 0, "pro": 1, "enterprise": 2}
async def check(user: User = Depends(get_current_user)):
if LEVELS.get(user.subscription, 0) < LEVELS.get(min_level, 0):
raise HTTPException(
status_code=403,
detail=f"Requires {min_level} plan. Current: {user.subscription}",
)
return user
return check
输出：
TEXT
📖 仅展示
# 函数定义成功
(3) 权限校验流程
flowchart LR
Request[Request with Token] --> ParseToken[Parse JWT Token]
ParseToken --> ValidToken{Token Valid?}
ValidToken -->|No| Return401[Return 401]
ValidToken -->|Yes| QueryUser[Query User from DB]
QueryUser --> UserExists{User Found?}
UserExists -->|No| Return401
UserExists -->|Yes| CheckSub{Subscription Level?}
CheckSub -->|Free| AllowBasic[Allow Basic Endpoints]
CheckSub -->|Pro| AllowPro[Allow Pro Endpoints]
CheckSub -->|Enterprise| AllowAll[Allow All Endpoints]
▶ 示例：受保护的商品路由
PYTHON
from fastapi import APIRouter, Depends, HTTPException, Path, Query
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional
from app.core.deps import get_db, get_current_user, require_subscription
from app.schemas import ProductCreate, ProductResponse, ProductUpdate
from app.services import ProductService
router = APIRouter()
@router.get("/products", response_model=list[ProductResponse])
async def list_products(
category: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
user=Depends(get_current_user),
):
service = ProductService(db)
return await service.list_products(category=category, skip=skip, limit=limit)
@router.post("/products", response_model=ProductResponse, status_code=201)
async def create_product(
product: ProductCreate,
db: AsyncSession = Depends(get_db),
user=Depends(get_current_user),
):
service = ProductService(db)
return await service.create_product(product, user_id=user.id)
@router.get("/products/{product_id}", response_model=ProductResponse)
async def get_product(
product_id: int = Path(gt=0),
db: AsyncSession = Depends(get_db),
user=Depends(get_current_user),
):
service = ProductService(db)
product = await service.get_product(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：批量导入端点（含订阅限制）
PYTHON
@router.post("/products/{product_id}/prices/bulk")
async def bulk_import_prices(
product_id: int = Path(gt=0),
prices: list[PriceCreate] = ...,
db: AsyncSession = Depends(get_db),
user=Depends(require_subscription("free")), # Even free users can import
):
# Free plan: max 1000 prices per import
limits = {"free": 1000, "pro": 100000, "enterprise": 1000000}
max_import = limits.get(user.subscription, 1000)
if len(prices) > max_import:
raise HTTPException(
status_code=403,
detail=f"Import limit: {max_import} for {user.subscription} plan. Upgrade at /pricing",
)
service = PriceService(db)
return await service.bulk_import_prices(product_id, prices)
输出：
TEXT
📖 仅展示
# 函数定义成功
❓ 常见问题
Q 中间件注册顺序为什么 CORS 要最后？
A CORS 需要给所有响应（包括 4xx 错误）添加 CORS 头。如果 CORS 不是最外层，限流中间件返回的 429 响应可能缺少 CORS 头，前端无法读取错误信息。
Q 路由组依赖和端点依赖会重复执行吗？
A 不会。FastAPI 对同一依赖做请求级缓存。路由组的
Depends(get_current_user) 和端点的 Depends(get_current_user) 共享同一次执行结果。Q DI 链中 get_db 被多次声明会有多个连接吗？
A 不会。同请求内 get_db 只执行一次，所有依赖共享同一个 AsyncSession。
Q Free 用户如何升级到 Pro？
A Stripe 支付集成（超出本课范围）。这里用 User.subscription 字段模拟，生产环境由支付回调更新。
Q 如何测试完整的请求链路？
A 用 TestClient + dependency_overrides 替换 get_db，逐步测试：无 Token → 401，Free Token → 200 但受限，Pro Token → 完整权限。
Q Phase 2 完成后下一步做什么？
A Phase 3 引入高级特性——WebSocket 实时推送、Celery 异步任务、Redis 缓存、文件上传、pytest 测试、OpenAPI 定制。
📖 小节
- Phase 2 整合了中间件、DI、数据库、CRUD、JWT 五大模块为完整可运行服务
- 三层 DI 链（get_db → get_current_user → require_subscription）统一管理认证和权限
- 路由分组策略：public 无认证，api_v1 JWT 保护，admin JWT + Admin 权限
- CORS 必须最外层注册，确保所有响应（包括错误）都有 CORS 头
- 订阅级别通过 DI 链实现：Free 限制 1000 条导入，Pro 100000 条，Enterprise 1000000 条
📝 作业
- 基础题（难度⭐）：将 Phase 1 的内存存储 CRUD 替换为异步 SQLAlchemy 实现，确保所有端点通过
Depends(get_db)注入数据库会话。提示：Repository 模式 +AsyncSession - 进阶题（难度⭐⭐）：实现路由分组——public（/health, /auth/login）无认证，api/v1 所有业务端点 JWT 保护，用 Swagger UI 测试登录+访问受保护端点。提示：
APIRouter(dependencies=[Depends(get_current_user)]) - 挑战题（难度⭐⭐⭐）：实现完整的三层 DI 链 + 批量导入端点：Pro 用户可导入 100000 条价格，Free 用户超 1000 条返回 403，同时在 Swagger UI 完整演示注册→登录→导入流程。提示：
require_subscription("free")+ limits 字典
---|
