FastAPI: 项目开发 — PriceTracker 从蓝图到代码

Q: mypy strict 模式值得吗？

值得。Strict 模式捕获更多类型错误，虽然初期配置成本高，但长期减少运行时 Bug。FastAPI + Pydantic 的类型提示天然适合 strict 模式。

Q: 如何验证"百万商品+千级并发"？

用 locust/k6 跑负载测试。先准备百万条测试数据（ INSERT INTO products SELECT generate_series(1, 1000000), ... ），再模拟千级并发查询。

Q: 代码审查流程怎么设计？

Bob/Alice 提交 PR → CI 自动跑 lint+test → Charlie 代码审查 → 合并到 develop → 自动部署到 staging → 验证后合并到 main。

最后更新：2026-08-26

项目开发像盖楼——先打地基（基础设施），再搭框架（认证+商品），然后填砖加瓦（价格+通知），最后精装修（报表+错误处理）。顺序错了，返工成本翻倍。

1. 你将学到

模块化开发顺序：基础设施 → 认证 → 商品 → 价格 → 通知 → 报表
异步优先原则：全链路 async/await，从路由到数据库
错误处理体系：自定义异常类 + 全局异常处理器 + 一致化错误响应格式
代码质量保障：pre-commit hooks + mypy + ruff + pytest 持续校验
Alice 场景交付验证：PriceTracker 可处理百万商品、千级并发、秒级价格推送

2. Alice 的真实故事

(1) 痛点：开发顺序混乱导致返工

Alice 先开发了价格导入功能，后来发现需要认证和权限检查，又回去改了 15 个端点。再后来发现错误响应格式不统一（有的返回 {"error": "..."}，有的返回 {"detail": "..."}），Bob 前端解析逻辑要写两套。开发顺序混乱导致 30% 的时间在返工。

(2) 模块化开发的解法

按依赖关系确定开发顺序：基础设施（DB/Redis/Config）→ 认证（JWT/权限）→ 商品（CRUD）→ 价格（导入/推送）→ 通知 → 报表。每个模块完成后测试通过再开始下一个，绝不返工。

(3) 收益

开发返工率从 30% 降到 5%，错误响应格式统一（所有错误返回 {"error": {"code": "...", "message": "..."}}），Bob 前端只需一套解析逻辑。

3. 模块开发顺序

(1) 依赖关系图

graph TD
    Infra[Infrastructure: DB/Redis/Config] --> Auth[Auth: JWT/Permissions]
    Auth --> Products[Products: CRUD]
    Products --> Prices[Prices: Import/Push]
    Auth --> Subscriptions[Subscriptions: Plans]
    Prices --> Notifications[Notifications: Alerts]
    Subscriptions --> Notifications
    Products --> Reports[Reports: Analytics]
    Prices --> Reports

顺序	模块	依赖	预计工时
1	基础设施	无	1 天
2	认证	基础设施	1 天
3	商品 CRUD	认证	1 天
4	价格导入/查询	商品	1 天
5	订阅计划	认证	0.5 天
6	通知告警	价格+订阅	1 天
7	报表统计	商品+价格	0.5 天

4. 错误处理体系

(1) 自定义异常类

▶ 示例：PriceTracker 异常体系

PYTHON

# app/core/exceptions.py
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

class PriceTrackerError(Exception):
    """Base exception for PriceTracker"""
    def __init__(self, code: str, message: str, status_code: int = 500):
        self.code = code
        self.message = message
        self.status_code = status_code

class NotFoundError(PriceTrackerError):
    def __init__(self, resource: str, resource_id: int | str):
        super().__init__(
            code=f"{resource.upper()}_NOT_FOUND",
            message=f"{resource} with id '{resource_id}' not found",
            status_code=404,
        )

class SubscriptionRequiredError(PriceTrackerError):
    def __init__(self, required_plan: str, current_plan: str):
        super().__init__(
            code="SUBSCRIPTION_REQUIRED",
            message=f"This feature requires {required_plan} plan. Current: {current_plan}",
            status_code=403,
        )

class ImportLimitError(PriceTrackerError):
    def __init__(self, limit: int, plan: str):
        super().__init__(
            code="IMPORT_LIMIT_EXCEEDED",
            message=f"Import limit: {limit} records for {plan} plan",
            status_code=403,
        )

输出：

TEXT 📖 仅展示

# 函数定义成功

(2) 全局异常处理器

flowchart TD
    A[Exception Raised] --> B{Exception Type?}
    B -->|PriceTrackerError| C[Format standard response]
    B -->|RequestValidationError| D[Format validation response]
    B -->|HTTPException| E[Format HTTP response]
    B -->|Other| F[Format 500 response]
    
    C --> G[JSONResponse: code + message]
    D --> G
    E --> G
    F --> G

▶ 示例：全局异常处理器注册

PYTHON

# app/core/exception_handlers.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from app.core.exceptions import PriceTrackerError

def format_error_response(code: str, message: str, status_code: int, details=None):
    return JSONResponse(
        status_code=status_code,
        content={
            "error": {
                "code": code,
                "message": message,
                "details": details,
            }
        },
    )

async def pricetracker_error_handler(request: Request, exc: PriceTrackerError):
    return format_error_response(exc.code, exc.message, exc.status_code)

async def validation_error_handler(request: Request, exc: RequestValidationError):
    return format_error_response(
        code="VALIDATION_ERROR",
        message="Request validation failed",
        status_code=422,
        details=exc.errors(),
    )

async def http_error_handler(request: Request, exc: HTTPException):
    return format_error_response(
        code="HTTP_ERROR",
        message=str(exc.detail),
        status_code=exc.status_code,
    )

# Register handlers
def register_exception_handlers(app: FastAPI):
    app.add_exception_handler(PriceTrackerError, pricetracker_error_handler)
    app.add_exception_handler(RequestValidationError, validation_error_handler)
    app.add_exception_handler(HTTPException, http_error_handler)

输出：

TEXT 📖 仅展示

# 函数定义成功

5. 异步优先原则

(1) 全链路 async 检查清单

层级	同步 ❌	异步 ✅
路由	`def get_xxx()`	`async def get_xxx()`
数据库	`Session` + `session.execute()`	`AsyncSession` + `await session.execute()`
HTTP 客户端	`requests.get()`	`httpx.AsyncClient().get()`
Redis	`redis.Redis()`	`redis.asyncio.Redis()`
文件 I/O	`open()` + `read()`	`aiofiles.open()` + `await read()`
任务	直接执行	Celery 异步任务

▶ 示例：全异步端点实现

PYTHON

# All layers are async
@app.get("/api/v1/products/{product_id}", response_model=ProductDetailResponse)
async def get_product_detail(
    product_id: int = Path(gt=0),
    db: AsyncSession = Depends(get_db),      # Async DB
    redis: Redis = Depends(get_redis),         # Async Redis
    user: User = Depends(get_current_user),     # Async auth
):
    # 1. Check cache (async)
    cached = await redis.get(f"product:{product_id}")
    if cached:
        return json.loads(cached)

    # 2. Query DB with eager loading (async)
    stmt = (
        select(Product)
        .options(selectinload(Product.prices))
        .where(Product.id == product_id)
    )
    result = await db.execute(stmt)
    product = result.scalar_one_or_none()

    if not product:
        raise NotFoundError("product", product_id)

    # 3. Cache result (async)
    data = ProductDetailResponse.model_validate(product).model_dump()
    await redis.setex(f"product:{product_id}", 300, json.dumps(data))

    return data

输出：

TEXT 📖 仅展示

# 函数定义成功

6. 代码质量保障

(1) 工具链配置

▶ 示例：pyproject.toml 质量工具配置

TOML

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "SIM"]

[tool.mypy]
python_version = "3.12"
strict = true
plugins = ["pydantic.mypy"]

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

输出：

TEXT 📖 仅展示

CI/CD 流水线配置已加载
Pipeline 运行状态: passed
Tests: 12 passed, 0 failed

▶ 示例：pre-commit 配置

YAML

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.5.0
    hooks:
      - id: ruff
        args: [--fix]
      - id: ruff-format

  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.10.0
    hooks:
      - id: mypy
        additional_dependencies: [pydantic, sqlalchemy, fastapi]

输出：

TEXT 📖 仅展示

CI/CD pipeline loaded
Pipeline status: passed
Tests: 12 passed, 0 failed

(2) 质量工具对比

工具	用途	运行时机
ruff	lint + format	pre-commit + CI
mypy	类型检查	pre-commit + CI
pytest	测试	CI + 手动
safety	依赖安全扫描	CI
bandit	代码安全检查	CI

❓ 常见问题

Q 开发顺序真的很重要吗？

A 非常重要。先开发依赖模块，后开发的模块可以直接使用已完成的功能，避免返工。基础设施→认证→业务是铁律。

Q 自定义异常和 HTTPException 怎么选？

A 业务错误用自定义异常（含 code + message），HTTP 层面错误用 HTTPException。自定义异常通过全局处理器统一格式化。

Q pre-commit 会不会太慢？

A ruff 极快（Rust 实现），mypy 稍慢但增量检查很快。总计 < 5 秒，换来的是每次提交都保证代码质量，远比 CI 失败后再修高效。

Q mypy strict 模式值得吗？

A 值得。Strict 模式捕获更多类型错误，虽然初期配置成本高，但长期减少运行时 Bug。FastAPI + Pydantic 的类型提示天然适合 strict 模式。

Q 如何验证"百万商品+千级并发"？

A 用 locust/k6 跑负载测试。先准备百万条测试数据（INSERT INTO products SELECT generate_series(1, 1000000), ...），再模拟千级并发查询。

Q 代码审查流程怎么设计？

A Bob/Alice 提交 PR → CI 自动跑 lint+test → Charlie 代码审查 → 合并到 develop → 自动部署到 staging → 验证后合并到 main。

📖 小节

模块开发顺序按依赖关系：基础设施→认证→商品→价格→通知→报表，避免返工
自定义异常体系 + 全局处理器实现统一错误响应格式：{"error": {"code": "...", "message": "..."}}
异步优先原则：全链路 async/await，同步代码用 run_in_executor 包装
代码质量保障：ruff（lint+format）+ mypy（类型检查）+ pytest（测试）+ pre-commit（自动校验）
PriceTracker 交付验证：百万商品、千级并发、秒级推送、统一错误格式

📝 作业

基础题（难度⭐）：实现 PriceTracker 的自定义异常类体系（PriceTrackerError → NotFoundError → SubscriptionRequiredError），在端点中用 raise NotFoundError("product", 42) 替代 raise HTTPException(404)。提示：继承 Exception
进阶题（难度⭐⭐）：实现全局异常处理器，注册 PriceTrackerError、RequestValidationError、HTTPException 三种处理器，确保所有错误响应格式统一。提示：app.add_exception_handler() + format_error_response()
挑战题（难度⭐⭐⭐）：配置完整的代码质量工具链——pyproject.toml（ruff + mypy + pytest 配置）、.pre-commit-config.yaml、运行 ruff check + mypy app/ + pytest 全部通过。提示：uv add --dev ruff mypy pytest + pre-commit install

---|

Web-Tutorial 技术团队

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

关于我们

FastAPI: 项目开发 — PriceTracker 从蓝图到代码

1. 你将学到

2. Alice 的真实故事

(1) 痛点：开发顺序混乱导致返工

(2) 模块化开发的解法

(3) 收益

3. 模块开发顺序

(1) 依赖关系图

4. 错误处理体系

(1) 自定义异常类

▶ 示例：PriceTracker 异常体系

(2) 全局异常处理器

▶ 示例：全局异常处理器注册

5. 异步优先原则

(1) 全链路 async 检查清单

▶ 示例：全异步端点实现

6. 代码质量保障

(1) 工具链配置

▶ 示例：pyproject.toml 质量工具配置

▶ 示例：pre-commit 配置

(2) 质量工具对比

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好