当当网书店购书中心实战：5步搞定API变更，附完整示例

版本升级后 API 全变了，是不是让你抓狂？别急，这其实是大多数开发者在维护老项目时的噩梦。今天我们就以当当网书店购书中心为原型，从零搭建一个高可用的后端服务，并给出一套应对 API 变更的完整示例。

项目目标与架构设计

我们要构建的是一个模拟当当网核心业务场景的购书中心。它不仅要处理商品查询、购物车、订单生成，还要应对真实世界中常见的“接口版本迭代”问题。

核心业务逻辑：

1. 商品检索：支持按书名、ISBN、分类进行模糊搜索。
2. 购物车管理：添加商品、修改数量、移除商品。
3. 订单结算：生成订单号，锁定库存，计算总价。
4. API 兼容性层：这是重点。当后端从 v1 升级到 v2 时，前端老版本代码不应崩溃。

技术选型：

• 语言：Python 3.9+
• 框架：FastAPI (异步高性能，自带文档生成，适合演示 API 变更)
• 数据库：SQLite (轻量级，便于本地运行完整示例)
• ORM：SQLAlchemy

为什么选 FastAPI？因为它对 Pydantic 模型的支持极好，非常适合处理数据验证和版本化。在 Stack Overflow 上，关于 FastAPI 版本控制的讨论非常多，官方推荐的方式是使用路由前缀或中间件拦截，我们将采用路由前缀+数据模型映射的混合策略，既简单又有效。

目录结构规划

清晰的目录结构是代码可维护性的基石。以下是我们项目的标准布局：

dangdang-book-center/
├── app/
│   ├── __init__.py
│   ├── main.py          # 应用入口
│   ├── database.py      # 数据库配置
│   ├── models.py        # 数据库模型 (ORM)
│   ├── schemas.py       # Pydantic 数据模式 (API 输入输出)
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   └── books.py # v1 版本 API
│   │   └── v2/
│   │       ├── __init__.py
│   │       └── books.py # v2 版本 API (模拟升级)
│   └── services/
│       ├── __init__.py
│       └── book_service.py # 核心业务逻辑
├── requirements.txt
└── README.md

注意 api 目录下分出了 v1 和 v2。这就是应对“API 全变了”的最直接手段：物理隔离。v1 保持向后兼容，v2 引入新特性（如增加“评分”字段、改变价格返回格式等）。

核心代码实现

1. 数据库模型与初始化

首先定义数据层。我们在 models.py 中定义 Book 和 Order 模型。

# app/models.py
from sqlalchemy import Column, Integer, String, Float, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetimeBase = declarative_base()class Book(Base):__tablename__ = 'books'id = Column(Integer, primary_key=True, index=True)isbn = Column(String, unique=True, index=True, nullable=False)title = Column(String, nullable=False)author = Column(String)price = Column(Float, nullable=False)# v2 新增字段，v1 不返回此字段rating = Column(Float, default=0.0) created_at = Column(DateTime, default=datetime.utcnow)class Order(Base):__tablename__ = 'orders'id = Column(Integer, primary_key=True, index=True)order_no = Column(String, unique=True, index=True, nullable=False)total_price = Column(Float, nullable=False)status = Column(String, default='PENDING')created_at = Column(DateTime, default=datetime.utcnow)

在 database.py 中初始化 SQLite：

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from .models import BaseSQLALCHEMY_DATABASE_URL = "sqlite:///./dangdang.db"engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)def init_db():Base.metadata.create_all(bind=engine)

2. Pydantic Schemas：定义 API 契约

这是处理 API 变更的关键。v1 和 v2 返回的数据结构不同，我们需要两套 Schema。

# app/schemas.py
from pydantic import BaseModel
from datetime import datetime# --- V1 Schemas ---
class BookOutV1(BaseModel):id: intisbn: strtitle: strauthor: strprice: float# 注意：这里没有 rating 字段class Config:orm_mode = True# --- V2 Schemas ---
class BookOutV2(BaseModel):id: intisbn: strtitle: strauthor: strprice: floatrating: float # v2 新增created_at: datetime # v2 新增class Config:orm_mode = True# 通用请求模型
class BookCreate(BaseModel):isbn: strtitle: strauthor: strprice: float

3. 业务逻辑服务层

将逻辑从路由中剥离，便于复用和测试。

# app/services/book_service.py
from sqlalchemy.orm import Session
from ..models import Bookdef get_books_by_query(db: Session, query: str):"""模糊搜索书籍"""return db.query(Book).filter(Book.title.ilike(f"%{query}%") | Book.isbn.ilike(f"%{query}%")).all()def get_book_by_id(db: Session, book_id: int):return db.query(Book).filter(Book.id == book_id).first()

4. API 路由实现：应对版本差异

V1 路由 (保持兼容)

# app/api/v1/books.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from ...database import SessionLocal
from ...models import Book
from ...schemas import BookOutV1
from ...services import book_servicerouter = APIRouter()def get_db():db = SessionLocal()try:yield dbfinally:db.close()@router.get("/books/{book_id}", response_model=BookOutV1)
def read_book(book_id: int, db: Session = Depends(get_db)):book = book_service.get_book_by_id(db, book_id)if book is None:raise HTTPException(status_code=404, detail="Book not found")return book

V2 路由 (新功能)

# app/api/v2/books.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from ...database import SessionLocal
from ...models import Book
from ...schemas import BookOutV2
from ...services import book_servicerouter = APIRouter()def get_db():db = SessionLocal()try:yield dbfinally:db.close()@router.get("/books/{book_id}", response_model=BookOutV2)
def read_book_v2(book_id: int, db: Session = Depends(get_db)):book = book_service.get_book_by_id(db, book_id)if book is None:raise HTTPException(status_code=404, detail="Book not found")return book@router.post("/books", response_model=BookOutV2)
def create_book(book_data, db: Session = Depends(get_db)):# 此处简化，实际需处理唯一约束冲突db_book = Book(**book_data.dict())db.add(db_book)db.commit()db.refresh(db_book)return db_book

5. 主应用入口：挂载不同版本

在 main.py 中，我们将不同版本的路由挂载到不同的前缀下。

# app/main.py
from fastapi import FastAPI
from .database import init_db, SessionLocal
from .models import Book
from .api.v1 import books as books_v1
from .api.v2 import books as books_v2app = FastAPI(title="当当网书店购书中心", description="模拟API版本升级的完整示例")@app.on_event("startup")
def on_startup():init_db()# 初始化一些测试数据db = SessionLocal()if not db.query(Book).first():sample_book = Book(isbn="978711545678", title="Python编程：从入门到实践", author="Eric Matthes", price=59.0, rating=4.8)db.add(sample_book)db.commit()db.close()# 挂载 v1
app.include_router(books_v1.router, prefix="/api/v1", tags=["Books-V1"])
# 挂载 v2
app.include_router(books_v2.router, prefix="/api/v2", tags=["Books-V2"])@app.get("/")
def root():return {"message": "Welcome to Dangdang Book Center API", "docs": "/docs"}

运行与测试

1. 环境准备

创建虚拟环境并安装依赖：

pip install fastapi uvicorn sqlalchemy pydantic

2. 启动服务

uvicorn app.main:app --reload

访问 http://127.0.0.1:8000/docs 查看自动生成的 Swagger 文档。

3. 测试 API 变更

测试 V1 接口：

curl -X GET "http://127.0.0.1:8000/api/v1/books/1"

预期响应：

{"id": 1,"isbn": "978711545678","title": "Python编程：从入门到实践","author": "Eric Matthes","price": 59.0
}

注意：这里没有 rating 和 created_at 字段，保持了向后兼容。

测试 V2 接口：

curl -X GET "http://127.0.0.1:8000/api/v2/books/1"

预期响应：

{"id": 1,"isbn": "978711545678","title": "Python编程：从入门到实践","author": "Eric Matthes","price": 59.0,"rating": 4.8,"created_at": "2023-10-27T10:00:00"
}

V2 接口返回了更丰富的数据。

进阶测试：创建新书 使用 V2 接口创建书籍：

curl -X POST "http://127.0.0.1:8000/api/v2/books" \-H "Content-Type: application/json" \-d '{"isbn": "978711556789","title": "Go 语言实战","author": "Bill Kennedy","price": 69.0}'

优化扩展与避坑指南

1. 为什么不用中间件拦截？

有些开发者喜欢用中间件检查请求头 X-API-Version，然后动态加载不同的 Schema。这种方式灵活，但调试困难。在 Stack Overflow 的高赞回答中，多数专家建议：对于重大版本变更，物理分离路由是最稳妥的做法。中间件更适合处理微小的、非破坏性的变更（如增加一个可选字段）。

2. 数据迁移问题

当从 V1 升级到 V2 时，数据库结构变了（增加了 rating 列）。

• 生产环境建议：使用 Alembic 进行数据库迁移。
• 本示例简化：SQLite 支持 ALTER TABLE ADD COLUMN，我们可以手动执行：

  ALTER TABLE books ADD COLUMN rating FLOAT DEFAULT 0.0;
  

  务必在升级前备份数据库！

3. 性能优化

• 缓存：书籍信息变化频率低，适合使用 Redis 缓存 V1/V2 的查询结果。
• 索引：确保 isbn 和 title 有索引，否则模糊搜索在大数据量下会极慢。

4. 常见坑点

• Pydantic 版本：确保使用 Pydantic v1 或 v2，API 略有不同。本示例基于 v1 的 orm_mode，v2 中改为 from_attributes = True。
• SQLite 并发：SQLite 是文件型数据库，高并发下写锁冲突严重。生产环境请替换为 PostgreSQL 或 MySQL。
• 日期序列化：FastAPI 自动处理 datetime 转 ISO 8601 字符串，但如果你的前端期望时间戳，需在 Schema 中自定义序列化器。

小结

通过本实战项目，我们不仅搭建了一个当当网书店购书中心的核心后端，更掌握了应对“版本升级后 API 全变了”的工程化思路。

核心要点回顾：

1. 物理隔离路由：/api/v1 和 /api/v2 分开，避免耦合。
2. Schema 分离：不同版本使用不同的 Pydantic 模型，确保数据结构可控。
3. 业务逻辑复用：Service 层不依赖具体 API 版本，只依赖数据库模型。

这套方案在业界非常通用，无论是电商、金融还是 SaaS 平台，处理 API 演进时都能直接套用。你不需要每次都重写整个后端，只需要新增一个版本目录，挂载新路由，然后逐步引导客户端迁移即可。

这个知识点你面试被问过吗？留言说说