一、项目背景

经过前 15 章的学习，星云电商团队已经掌握了 SQLAlchemy 2.0 的核心知识：从 Engine/Connection 的连接管理，到 MetaData 的 Schema 声明，再到 ORM 的关系映射、事务隔离、乐观并发控制、Alembic 迁移和 pytest 测试基建。现在是时候将这些知识融会贯通，交付一个完整的、可运行的最小可行产品（MVP）。

本综合实战的场景设定为"618 大促前的订单中台改造"。星云电商原先的订单系统是一套用裸 psycopg2 + SQL 字符串拼接的 PHP 服务，经历多年迭代后已经成了无人敢碰的"代码坟场"。架构团队决定用 Python + SQLAlchemy 2.0 重建订单中台核心模块，要求在 618 大促前完成以下功能上线：

1. 用户下单（带库存扣减和乐观并发控制）
2. 订单明细查询（我的订单列表，支持分页和筛选）
3. 库存管理（商品库存查询、补货）
4. Schema 迁移（全部通过 Alembic 管理，不允许手动 ALTER TABLE）
5. 自动化测试（核心用例覆盖，CI 全绿）

本章将以单一项目仓库的形式，串联前 15 章的每一个知识点，输出一个可直接运行、带接口、带测试、带迁移的"星云订单中台 MVP"。读者可以在本章之后直接基于这个 MVP 添砖加瓦，构筑自己的业务系统。

二、项目设计

场景：周一上午，大会议室。大师召集了全体技术人员，宣布启动"订单中台 MVP"项目。白板上画着完整的架构图和里程碑。

大师：“618 只剩六周。我们的 PHP 订单系统每到大促就当机。这次我们用 SQLAlchemy 2.0 重写订单核心。小胖负责模型设计和关系映射，小白负责事务和测试，我负责架构把关和 Alembic 迁移流程。目标是——两周内 MVP 上线。”

小胖：“两周？我们连模型设计都还没开始！而且 PHP 那边的订单表有 30 多个字段，关系复杂得一塌糊涂……”

大师：“MVP 不是全功能上线，而是’核心链路可跑’。我们只做四条核心链路：用户注册、商品列表、下单（扣库存+创建订单+明细）、我的订单。功能少，但每一条链路都要做到——事务安全、并发安全、迁移可回滚、测试全覆盖。”

小白：“那技术选型呢？FastAPI 还是 Flask？我看大纲里说’FastAPI/Flask 任选’。”

大师：“FastAPI。因为第 17 章我们就要讲到异步了，这个 MVP 先用同步，但目录结构和依赖注入方式要和异步模式兼容。所以我们用 FastAPI 的同步依赖注入——实际上 FastAPI 的 DI 本就是异步的，但 SQLAlchemy 的同步 Session 也能正常使用。”

小胖：“技术映射：FastAPI 依赖注入 = 食堂的分餐窗口，每个请求自动配餐（注入 Session）。那目录结构按第 2 章的来？”

大师：“对，基于第 2 章的骨架，我们在 src/order_center/ 下补充几个新模块：”

nebula-order-center/
├── pyproject.toml
├── docker-compose.yml
├── alembic/
│   ├── env.py
│   ├── versions/
│   └── alembic.ini
├── src/
│   └── order_center/
│       ├── __init__.py
│       ├── config.py
│       ├── db/
│       │   ├── __init__.py
│       │   ├── engine.py          # Engine 单例
│       │   └── session.py         # Session 工厂
│       ├── models/
│       │   ├── __init__.py
│       │   ├── base.py            # DeclarativeBase
│       │   ├── user.py            # User
│       │   ├── product.py         # Product（带 version_id）
│       │   └── order.py           # Order + OrderItem
│       ├── repository/            # 数据访问层（新增）
│       │   ├── __init__.py
│       │   ├── user_repo.py
│       │   ├── product_repo.py
│       │   └── order_repo.py
│       ├── service/               # 业务逻辑层（新增）
│       │   ├── __init__.py
│       │   ├── order_service.py
│       │   └── product_service.py
│       └── api/                   # FastAPI 路由（新增）
│           ├── __init__.py
│           ├── main.py
│           ├── deps.py            # 依赖注入
│           └── routes/
│               ├── orders.py
│               └── products.py
├── tests/
│   ├── conftest.py
│   ├── factories/
│   │   └── order_factories.py
│   ├── test_models/
│   ├── test_repository/
│   ├── test_service/
│   └── test_api/
└── scripts/
    └── health_check.py

小白：“Repository 层之前没有专门讲过。它的职责边界是什么？”

大师：“Repository 是数据访问的抽象层——它封装了所有与数据库交互的代码（构造查询、执行查询、返回结果）。Service 层不关心数据从哪来、怎么查的，只关心业务逻辑。这种分层的好处是——当查询逻辑需要优化（比如第 19 章改变加载策略），只改 Repository，Service 和 API 都不受影响。”

小胖：“技术映射：Repository = 数据仓库管理员（你告诉他要什么，他去库房取）；Service = 业务经理（决定怎么处理这些数据）。”

小白：“那事务边界呢？Service 层还是 Repository 层管理事务？”

大师：“事务边界在 Service 层。Repository 只负责构造和执行查询，不管事务的 begin/commit/rollback。Service 层调用 session.commit()。这样确保一个业务操作（如’下单’——涉及 Product 扣减、Order 创建、OrderItem 创建）在同一个事务中完成。”

小胖：“还有一个问题：我们怎么验证 MVP 的正确性？总不能上线后发现库存超卖吧？”

大师：“三层验证。第一层：Repository 层的单元测试——验证每个查询方法返回正确的数据。第二层：Service 层的集成测试——验证业务逻辑的正确性（如库存不足抛异常、乐观锁冲突重试）。第三层：API 层的端到端测试——用 FastAPI 的 TestClient 走完整 HTTP 流程。每一层都通过事务回滚隔离（第 15 章）。”

小白：“技术映射：三层测试金字塔 = 单元→集成→端到端，从底层推到顶层。那 Alembic 迁移呢？MVP 开发过程中 Schema 会频繁变更。”

大师："开发流程是：

修改模型 → alembic autogenerate → review 生成的脚本 → alembic upgrade head → 运行测试

每次修改模型后，生成一条新的 migration。决不在生产库上手动 ALTER TABLE。所有的 Schema 变更必须通过 Alembic 完成——这是团队的铁律。"

小胖：“明白了。两周后见分晓！”

三、项目实战

实战目标

构建"星云订单中台 MVP"完整项目，包含：

• 4 个模型（User / Product / Order / OrderItem）
• Repository 层（封装数据访问）
• Service 层（事务边界 + 乐观并发锁 + 重试逻辑）
• FastAPI 路由（用户注册、商品查询、下单、订单列表）
• Alembic 迁移脚本
• pytest 测试（单元 + 集成 + API）
• Docker Compose 一键启动

步骤一：模型层（models/）

# src/order_center/models/base.py
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import MetaData

naming_convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=naming_convention)

# src/order_center/models/user.py
from sqlalchemy import String, Integer, Numeric, DateTime, Boolean, text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from datetime import datetime
from typing import Optional, List
from order_center.models.base import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, index=True)
    email: Mapped[str] = mapped_column(String(120), unique=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
    phone: Mapped[Optional[str]] = mapped_column(String(20), nullable=True)
    real_name: Mapped[Optional[str]] = mapped_column(String(50), nullable=True)
    balance: Mapped[float] = mapped_column(Numeric(10, 2), server_default=text("0.00"))
    is_active: Mapped[bool] = mapped_column(Boolean, server_default=text("TRUE"))
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

    orders: Mapped[List["Order"]] = relationship(back_populates="user")

    def __repr__(self):
        return f"<User(id={self.id}, username={self.username})>"

# src/order_center/models/product.py
from sqlalchemy import String, Integer, Numeric, DateTime, Boolean, Text, CheckConstraint, text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from datetime import datetime
from typing import Optional, List
from order_center.models.base import Base

class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    sku: Mapped[str] = mapped_column(String(30), unique=True, nullable=False, index=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    unit_price: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False)
    inventory: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
    category: Mapped[str] = mapped_column(String(50), nullable=False, index=True)
    status: Mapped[str] = mapped_column(String(20), nullable=False, server_default=text("'online'"))
    # 乐观并发控制
    version_id: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("1"))
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

    __mapper_args__ = {"version_id_col": version_id}
    __table_args__ = (
        CheckConstraint("unit_price > 0", name="ck_product_price_positive"),
        CheckConstraint("inventory >= 0", name="ck_product_inventory_non_negative"),
    )

    def __repr__(self):
        return f"<Product({self.sku}, {self.title}, ¥{self.unit_price})>"

# src/order_center/models/order.py
from sqlalchemy import String, Integer, Numeric, DateTime, ForeignKey, Index, UniqueConstraint, text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from datetime import datetime
from typing import Optional, List
from order_center.models.base import Base

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    order_no: Mapped[str] = mapped_column(String(32), unique=True, nullable=False, index=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False, index=True)
    total_amount: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False)
    status: Mapped[str] = mapped_column(String(20), nullable=False, server_default=text("'pending'"), index=True)
    payment_method: Mapped[Optional[str]] = mapped_column(String(20), nullable=True)
    idempotency_key: Mapped[Optional[str]] = mapped_column(String(64), unique=True, nullable=True)
    remark: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

    user: Mapped["User"] = relationship(back_populates="orders")
    items: Mapped[List["OrderItem"]] = relationship(back_populates="order", cascade="all, delete-orphan")

    __table_args__ = (
        Index("ix_orders_user_status", "user_id", "status"),
        Index("ix_orders_created", "created_at"),
    )

    def __repr__(self):
        return f"<Order({self.order_no}, ¥{self.total_amount}, {self.status})>"

class OrderItem(Base):
    __tablename__ = "order_items"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"), nullable=False, index=True)
    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"), nullable=False)
    product_name: Mapped[str] = mapped_column(String(200), nullable=False)
    unit_price: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False)
    quantity: Mapped[int] = mapped_column(Integer, nullable=False)
    subtotal: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

    order: Mapped["Order"] = relationship(back_populates="items")
    product: Mapped["Product"] = relationship()

    def __repr__(self):
        return f"<OrderItem({self.product_name} x{self.quantity}, ¥{self.subtotal})>"

# src/order_center/models/__init__.py
from order_center.models.base import Base
from order_center.models.user import User
from order_center.models.product import Product
from order_center.models.order import Order, OrderItem

__all__ = ["Base", "User", "Product", "Order", "OrderItem"]

步骤二：配置与 DB 层

# src/order_center/config.py
import os

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+psycopg://nebula:nebula_dev@localhost:5432/order_center"
)
DATABASE_URL_TEST = os.getenv(
    "DATABASE_URL_TEST",
    "postgresql+psycopg://nebula:nebula_dev@localhost:5432/order_center_test"
)
POOL_SIZE = int(os.getenv("POOL_SIZE", "5"))
MAX_OVERFLOW = int(os.getenv("MAX_OVERFLOW", "10"))
POOL_TIMEOUT = int(os.getenv("POOL_TIMEOUT", "30"))
POOL_RECYCLE = int(os.getenv("POOL_RECYCLE", "1800"))
ECHO_SQL = os.getenv("ECHO_SQL", "false").lower() == "true"

# src/order_center/db/engine.py
from sqlalchemy import create_engine
from order_center.config import DATABASE_URL, POOL_SIZE, MAX_OVERFLOW, POOL_TIMEOUT, POOL_RECYCLE, ECHO_SQL

engine = create_engine(
    DATABASE_URL,
    echo=ECHO_SQL,
    pool_size=POOL_SIZE,
    max_overflow=MAX_OVERFLOW,
    pool_timeout=POOL_TIMEOUT,
    pool_recycle=POOL_RECYCLE,
    pool_pre_ping=True,
)

# src/order_center/db/session.py
from sqlalchemy.orm import sessionmaker, Session
from order_center.db.engine import engine

SessionFactory = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
    expire_on_commit=True,
)

def get_session() -> Session:
    return SessionFactory()

步骤三：Repository 层

# src/order_center/repository/user_repo.py
from sqlalchemy.orm import Session
from sqlalchemy import select, or_
from order_center.models import User

class UserRepository:
    def __init__(self, session: Session):
        self.session = session

    def get_by_id(self, user_id: int) -> User | None:
        return self.session.get(User, user_id)

    def get_by_username(self, username: str) -> User | None:
        stmt = select(User).where(User.username == username)
        return self.session.execute(stmt).scalars().first()

    def get_by_email(self, email: str) -> User | None:
        stmt = select(User).where(User.email == email)
        return self.session.execute(stmt).scalars().first()

    def exists_by_username_or_email(self, username: str, email: str) -> User | None:
        stmt = select(User).where(
            or_(User.username == username, User.email == email)
        )
        return self.session.execute(stmt).scalars().first()

    def create(self, user: User) -> User:
        self.session.add(user)
        self.session.flush()
        return user

# src/order_center/repository/product_repo.py
from sqlalchemy.orm import Session
from sqlalchemy import select, func, and_, or_
from order_center.models import Product

class ProductRepository:
    def __init__(self, session: Session):
        self.session = session

    def get_by_id(self, product_id: int) -> Product | None:
        return self.session.get(Product, product_id)

    def get_by_sku(self, sku: str) -> Product | None:
        return self.session.execute(
            select(Product).where(Product.sku == sku)
        ).scalars().first()

    def search(self, *, keyword: str = None, category: str = None,
               min_price: float = None, max_price: float = None,
               status: str = "online", sort_by: str = "created_at_desc",
               page: int = 1, page_size: int = 20) -> list[Product]:
        stmt = select(Product).where(Product.status == status)
        if keyword:
            stmt = stmt.where(Product.title.ilike(f"%{keyword}%"))
        if category:
            stmt = stmt.where(Product.category == category)
        if min_price is not None:
            stmt = stmt.where(Product.unit_price >= min_price)
        if max_price is not None:
            stmt = stmt.where(Product.unit_price <= max_price)
        sort_map = {
            "created_at_desc": Product.created_at.desc(),
            "price_asc": Product.unit_price.asc(),
            "price_desc": Product.unit_price.desc(),
        }
        stmt = stmt.order_by(sort_map.get(sort_by, Product.created_at.desc()))
        stmt = stmt.limit(page_size).offset((page - 1) * page_size)
        return list(self.session.execute(stmt).scalars().all())

    def count_search(self, **kwargs) -> int:
        kwargs.pop("sort_by", None)
        kwargs.pop("page", None)
        kwargs.pop("page_size", None)
        stmt = select(func.count()).select_from(Product).where(Product.status == kwargs.get("status", "online"))
        if kw := kwargs.get("keyword"):
            stmt = stmt.where(Product.title.ilike(f"%{kw}%"))
        if cat := kwargs.get("category"):
            stmt = stmt.where(Product.category == cat)
        return self.session.execute(stmt).scalar()

# src/order_center/repository/order_repo.py
from sqlalchemy.orm import Session
from sqlalchemy import select, func, and_, desc
from order_center.models import Order, OrderItem
from datetime import datetime

class OrderRepository:
    def __init__(self, session: Session):
        self.session = session

    def get_by_id(self, order_id: int) -> Order | None:
        return self.session.get(Order, order_id)

    def get_by_order_no(self, order_no: str) -> Order | None:
        return self.session.execute(
            select(Order).where(Order.order_no == order_no)
        ).scalars().first()

    def get_by_idempotency_key(self, key: str) -> Order | None:
        return self.session.execute(
            select(Order).where(Order.idempotency_key == key)
        ).scalars().first()

    def get_user_orders(self, user_id: int, *, status: str = None,
                        page: int = 1, page_size: int = 20) -> list[Order]:
        stmt = select(Order).where(Order.user_id == user_id)
        if status:
            stmt = stmt.where(Order.status == status)
        stmt = stmt.order_by(Order.created_at.desc())
        stmt = stmt.limit(page_size).offset((page - 1) * page_size)
        return list(self.session.execute(stmt).scalars().all())

    def count_user_orders(self, user_id: int, status: str = None) -> int:
        stmt = select(func.count()).select_from(Order).where(Order.user_id == user_id)
        if status:
            stmt = stmt.where(Order.status == status)
        return self.session.execute(stmt).scalar()

    def create(self, order: Order) -> Order:
        self.session.add(order)
        self.session.flush()
        return order

步骤四：Service 层

# src/order_center/service/order_service.py
"""下单服务——事务边界 + 乐观并发 + 幂等 + 重试"""

import hashlib
import time
from sqlalchemy.orm import Session
from sqlalchemy.exc import StaleDataError
from order_center.models import User, Product, Order, OrderItem
from order_center.repository.user_repo import UserRepository
from order_center.repository.product_repo import ProductRepository
from order_center.repository.order_repo import OrderRepository

class OrderService:
    def __init__(self, session: Session):
        self.session = session
        self.user_repo = UserRepository(session)
        self.product_repo = ProductRepository(session)
        self.order_repo = OrderRepository(session)

    def register_user(self, username: str, email: str, password: str,
                      phone: str = None, real_name: str = None) -> User:
        """用户注册——校验用户名/邮箱唯一性"""
        existing = self.user_repo.exists_by_username_or_email(username, email)
        if existing:
            if existing.username == username:
                raise ValueError(f"用户名 {username} 已被注册")
            if existing.email == email:
                raise ValueError(f"邮箱 {email} 已被注册")

        password_hash = hashlib.sha256(password.encode()).hexdigest()
        user = User(
            username=username, email=email, password_hash=password_hash,
            phone=phone, real_name=real_name,
        )
        self.user_repo.create(user)
        self.session.commit()
        return user

    def place_order(self, user_id: int, items_data: list[dict],
                    idempotency_key: str = None, remark: str = None,
                    max_retries: int = 3) -> Order:
        """
        下单：扣库存 + 创建订单 + 明细
        - items_data: [{"product_id": 1, "quantity": 2}, ...]
        - idempotency_key: 幂等键，防止重复提交
        """
        # 幂等检查
        if idempotency_key:
            existing = self.order_repo.get_by_idempotency_key(idempotency_key)
            if existing:
                return existing

        for attempt in range(1, max_retries + 1):
            try:
                return self._place_order_internal(user_id, items_data, idempotency_key, remark)
            except StaleDataError:
                self.session.rollback()
                if attempt == max_retries:
                    raise RuntimeError(f"下单失败：重试 {max_retries} 次后仍版本冲突")
                time.sleep(0.01 * attempt)
                # 重建 session 内的状态（rollback 后对象过期）
                continue
            except Exception:
                self.session.rollback()
                raise

        raise RuntimeError("下单失败：不可达")

    def _place_order_internal(self, user_id: int, items_data: list[dict],
                               idempotency_key: str, remark: str) -> Order:
        """内部下单逻辑——单次尝试"""
        # 1. 校验用户
        user = self.user_repo.get_by_id(user_id)
        if not user:
            raise ValueError(f"用户 {user_id} 不存在")

        # 2. 逐个校验商品库存并计算总金额
        products = []
        total_amount = 0.0
        for item in items_data:
            product = self.product_repo.get_by_id(item["product_id"])
            if not product or product.status != "online":
                raise ValueError(f"商品 {item['product_id']} 不可购买")
            if product.inventory < item["quantity"]:
                raise ValueError(f"商品 {product.title} 库存不足：需要 {item['quantity']}，剩余 {product.inventory}")
            products.append((product, item["quantity"]))
            total_amount += product.unit_price * item["quantity"]

        # 3. 扣减库存（乐观锁：版本号冲突时抛 StaleDataError）
        for product, qty in products:
            product.inventory -= qty
            # flush 时自动生成 UPDATE WHERE version_id = :current_version

        # 4. 创建订单
        import uuid as _uuid
        import random
        order = Order(
            order_no=f"ORD-{int(time.time())}-{random.randint(1000, 9999)}",
            user_id=user_id,
            total_amount=total_amount,
            status="paid",
            idempotency_key=idempotency_key,
            remark=remark,
        )
        self.order_repo.create(order)

        # 5. 创建订单明细
        for product, qty in products:
            item = OrderItem(
                order_id=order.id,
                product_id=product.id,
                product_name=product.title,
                unit_price=product.unit_price,
                quantity=qty,
                subtotal=product.unit_price * qty,
            )
            self.session.add(item)

        self.session.commit()
        return order

    def get_user_orders(self, user_id: int, status: str = None,
                        page: int = 1, page_size: int = 20) -> tuple[list[Order], int]:
        """查询用户订单列表（分页）"""
        orders = self.order_repo.get_user_orders(user_id, status=status, page=page, page_size=page_size)
        total = self.order_repo.count_user_orders(user_id, status=status)
        return orders, total

# src/order_center/service/product_service.py
"""商品服务"""

from sqlalchemy.orm import Session
from order_center.models import Product
from order_center.repository.product_repo import ProductRepository

class ProductService:
    def __init__(self, session: Session):
        self.session = session
        self.repo = ProductRepository(session)

    def search_products(self, **kwargs) -> tuple[list[Product], int]:
        products = self.repo.search(**kwargs)
        total = self.repo.count_search(**kwargs)
        return products, total

    def get_by_id(self, product_id: int) -> Product | None:
        return self.repo.get_by_id(product_id)

    def add_inventory(self, product_id: int, quantity: int) -> Product:
        product = self.repo.get_by_id(product_id)
        if not product:
            raise ValueError(f"商品 {product_id} 不存在")
        product.inventory += quantity
        self.session.commit()
        return product

步骤五：FastAPI 路由层

# src/order_center/api/deps.py
"""FastAPI 依赖注入"""

from fastapi import Depends
from sqlalchemy.orm import Session
from order_center.db.session import get_session
from order_center.service.order_service import OrderService
from order_center.service.product_service import ProductService

def get_db():
    """每个 HTTP 请求创建一个 Session，请求结束时 commit 或 rollback"""
    session = get_session()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

def get_order_service(session: Session = Depends(get_db)) -> OrderService:
    return OrderService(session)

def get_product_service(session: Session = Depends(get_db)) -> ProductService:
    return ProductService(session)

# src/order_center/api/main.py
from fastapi import FastAPI
from order_center.api.routes import orders, products

app = FastAPI(
    title="星云订单中台 MVP",
    version="0.1.0",
    description="SQLAlchemy 2.0 专栏综合实战——订单中台核心 API"
)

app.include_router(orders.router, prefix="/api/v1/orders", tags=["订单"])
app.include_router(products.router, prefix="/api/v1/products", tags=["商品"])

@app.get("/health")
def health_check():
    return {"status": "ok", "service": "order-center-mvp"}

# src/order_center/api/routes/orders.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
from order_center.api.deps import get_order_service
from order_center.service.order_service import OrderService
from order_center.models import Order, OrderItem

router = APIRouter()

# ---------- Pydantic Schemas ----------

class OrderItemRequest(BaseModel):
    product_id: int
    quantity: int = Field(ge=1, le=999)

class PlaceOrderRequest(BaseModel):
    user_id: int
    items: list[OrderItemRequest]
    idempotency_key: str | None = None
    remark: str | None = None

class RegisterUserRequest(BaseModel):
    username: str = Field(min_length=2, max_length=50)
    email: str = Field(max_length=120)
    password: str = Field(min_length=6)
    phone: str | None = None
    real_name: str | None = None

class OrderItemResponse(BaseModel):
    id: int
    product_id: int
    product_name: str
    unit_price: float
    quantity: int
    subtotal: float

    class Config:
        from_attributes = True

class OrderResponse(BaseModel):
    id: int
    order_no: str
    user_id: int
    total_amount: float
    status: str
    idempotency_key: str | None
    remark: str | None
    created_at: str
    items: list[OrderItemResponse] = []

    class Config:
        from_attributes = True

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    phone: str | None
    real_name: str | None
    created_at: str

    class Config:
        from_attributes = True

# ---------- Routes ----------

@router.post("/register", response_model=UserResponse)
def register_user(req: RegisterUserRequest, svc: OrderService = Depends(get_order_service)):
    """用户注册"""
    try:
        user = svc.register_user(
            username=req.username, email=req.email, password=req.password,
            phone=req.phone, real_name=req.real_name,
        )
        return user
    except ValueError as e:
        raise HTTPException(status_code=409, detail=str(e))

@router.post("/place", response_model=OrderResponse)
def place_order(req: PlaceOrderRequest, svc: OrderService = Depends(get_order_service)):
    """下单"""
    try:
        items_data = [{"product_id": i.product_id, "quantity": i.quantity} for i in req.items]
        order = svc.place_order(
            user_id=req.user_id, items_data=items_data,
            idempotency_key=req.idempotency_key, remark=req.remark,
        )
        return _order_to_response(order)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except RuntimeError as e:
        raise HTTPException(status_code=503, detail=str(e))

@router.get("/user/{user_id}", response_model=dict)
def get_user_orders(
    user_id: int,
    status: str = None,
    page: int = 1,
    page_size: int = 20,
    svc: OrderService = Depends(get_order_service),
):
    """查询用户订单列表"""
    orders, total = svc.get_user_orders(user_id, status=status, page=page, page_size=page_size)
    return {
        "data": [_order_to_response(o) for o in orders],
        "total": total,
        "page": page,
        "page_size": page_size,
    }

@router.get("/{order_id}", response_model=OrderResponse)
def get_order(order_id: int, svc: OrderService = Depends(get_order_service)):
    """查询订单详情"""
    order = svc.order_repo.get_by_id(order_id)
    if not order:
        raise HTTPException(status_code=404, detail="订单不存在")
    return _order_to_response(order)

def _order_to_response(order: Order) -> dict:
    return {
        "id": order.id,
        "order_no": order.order_no,
        "user_id": order.user_id,
        "total_amount": float(order.total_amount),
        "status": order.status,
        "idempotency_key": order.idempotency_key,
        "remark": order.remark,
        "created_at": order.created_at.isoformat() if order.created_at else None,
        "items": [
            {
                "id": item.id,
                "product_id": item.product_id,
                "product_name": item.product_name,
                "unit_price": float(item.unit_price),
                "quantity": item.quantity,
                "subtotal": float(item.subtotal),
            }
            for item in (order.items or [])
        ],
    }

# src/order_center/api/routes/products.py
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from typing import Optional
from order_center.api.deps import get_product_service
from order_center.service.product_service import ProductService

router = APIRouter()

class ProductResponse(BaseModel):
    id: int
    sku: str
    title: str
    unit_price: float
    inventory: int
    category: str
    status: str
    created_at: str

    class Config:
        from_attributes = True

@router.get("/search", response_model=dict)
def search_products(
    keyword: str = None,
    category: str = None,
    min_price: float = None,
    max_price: float = None,
    sort_by: str = "created_at_desc",
    page: int = 1,
    page_size: int = 20,
    svc: ProductService = Depends(get_product_service),
):
    """商品搜索"""
    products, total = svc.search_products(
        keyword=keyword, category=category,
        min_price=min_price, max_price=max_price,
        sort_by=sort_by, page=page, page_size=page_size,
    )
    return {
        "data": products,
        "total": total,
        "page": page,
        "page_size": page_size,
    }

@router.get("/{product_id}", response_model=ProductResponse)
def get_product(product_id: int, svc: ProductService = Depends(get_product_service)):
    """商品详情"""
    product = svc.get_by_id(product_id)
    if not product:
        raise HTTPException(status_code=404, detail="商品不存在")
    return product

步骤六：Alembic 迁移

# 1. 初始化 Alembic
alembic init alembic

# 2. 修改 alembic/env.py，导入 Base
# from order_center.models import Base
# target_metadata = Base.metadata

# 3. 修改 alembic.ini
# sqlalchemy.url = postgresql+psycopg://nebula:nebula_dev@localhost:5432/order_center

# 4. 生成初始迁移
alembic revision --autogenerate -m "initial schema"

# 5. 应用迁移
alembic upgrade head

# 6. 验证表结构
# python scripts/health_check.py

"""alembic/env.py 关键配置"""
from order_center.models import Base
target_metadata = Base.metadata

# 在线迁移模式
def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            compare_server_default=True,
        )
        with context.begin_transaction():
            context.run_migrations()

步骤七：测试层

# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from order_center.models import Base
from order_center.config import DATABASE_URL_TEST

@pytest.fixture(scope="session")
def engine():
    test_engine = create_engine(DATABASE_URL_TEST, echo=False, pool_size=3, max_overflow=5)
    Base.metadata.create_all(test_engine)
    return test_engine

@pytest.fixture
def connection(engine):
    conn = engine.connect()
    trans = conn.begin()
    yield conn
    trans.rollback()
    conn.close()

@pytest.fixture
def session(connection):
    Factory = sessionmaker(bind=connection, autocommit=False, autoflush=False)
    s = Factory()
    yield s
    s.close()

# 测试数据工厂
def create_test_user(session: Session, username="testuser") -> "User":
    from order_center.models import User
    user = User(username=username, email=f"{username}@test.com", password_hash="hash", phone="13900000000")
    session.add(user)
    session.flush()
    return user

def create_test_product(session: Session, sku="SKU-001", title="测试商品",
                         price=99.0, inventory=10) -> "Product":
    from order_center.models import Product
    product = Product(sku=sku, title=title, unit_price=price, inventory=inventory, category="测试")
    session.add(product)
    session.flush()
    return product

# tests/test_repository/test_user_repo.py
from order_center.repository.user_repo import UserRepository
from tests.conftest import create_test_user

def test_get_by_username(session):
    user = create_test_user(session, "alice")
    repo = UserRepository(session)
    found = repo.get_by_username("alice")
    assert found is not None
    assert found.username == "alice"

def test_exists_by_username_or_email(session):
    create_test_user(session, "bob")
    repo = UserRepository(session)
    existing = repo.exists_by_username_or_email("bob", "other@test.com")
    assert existing is not None
    not_found = repo.exists_by_username_or_email("nobody", "nobody@test.com")
    assert not_found is None

# tests/test_service/test_order_service.py
import pytest
from order_center.service.order_service import OrderService
from sqlalchemy.exc import StaleDataError
from tests.conftest import create_test_user, create_test_product

def test_place_order_success(session):
    user = create_test_user(session)
    product = create_test_product(session, inventory=10)
    svc = OrderService(session)

    order = svc.place_order(
        user_id=user.id,
        items_data=[{"product_id": product.id, "quantity": 3}],
    )
    assert order is not None
    assert order.status == "paid"
    assert len(order.items) == 1
    assert order.items[0].quantity == 3

    # 验证库存扣减
    session.refresh(product)
    assert product.inventory == 7

def test_place_order_insufficient_inventory(session):
    user = create_test_user(session)
    product = create_test_product(session, inventory=2)
    svc = OrderService(session)

    with pytest.raises(ValueError, match="库存不足"):
        svc.place_order(user_id=user.id, items_data=[{"product_id": product.id, "quantity": 5}])

def test_place_order_idempotent(session):
    user = create_test_user(session)
    product = create_test_product(session, inventory=10)
    svc = OrderService(session)
    idem_key = "idem-test-001"

    order1 = svc.place_order(user_id=user.id, items_data=[{"product_id": product.id, "quantity": 1}], idempotency_key=idem_key)
    order2 = svc.place_order(user_id=user.id, items_data=[{"product_id": product.id, "quantity": 1}], idempotency_key=idem_key)

    assert order1.id == order2.id  # 幂等：返回同一个订单

    # 库存只扣一次
    session.refresh(product)
    assert product.inventory == 9

def test_place_order_optimistic_lock(session):
    """验证乐观并发锁：并发修改时 StaleDataError"""
    user = create_test_user(session)
    product = create_test_product(session, inventory=10)
    # 模拟另一个事务修改了同一行
    from sqlalchemy import text
    session.execute(text("UPDATE products SET inventory = 5, version_id = version_id + 1 WHERE id = :id"), {"id": product.id})
    session.flush()
    # 现在 product 对象的版本号已过期
    svc = OrderService(session)
    with pytest.raises((StaleDataError, RuntimeError)):
        svc.place_order(user_id=user.id, items_data=[{"product_id": product.id, "quantity": 3}])

# tests/test_api/test_orders_api.py
from fastapi.testclient import TestClient
from order_center.api.main import app
from tests.conftest import create_test_user, create_test_product

client = TestClient(app)

def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "ok"

def test_register_user(session):
    response = client.post("/api/v1/orders/register", json={
        "username": "api_user",
        "email": "api@test.com",
        "password": "password123",
    })
    assert response.status_code == 200
    data = response.json()
    assert data["username"] == "api_user"

def test_place_order_api(session):
    user = create_test_user(session, "order_api_user")
    product = create_test_product(session, "SKU-API", inventory=20)
    response = client.post("/api/v1/orders/place", json={
        "user_id": user.id,
        "items": [{"product_id": product.id, "quantity": 2}],
    })
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "paid"
    assert len(data["items"]) == 1

def test_get_user_orders(session):
    user = create_test_user(session, "list_api_user")
    response = client.get(f"/api/v1/orders/user/{user.id}?page=1&page_size=10")
    assert response.status_code == 200
    data = response.json()
    assert "data" in data
    assert "total" in data

步骤八：启动运行

# 1. 启动数据库
docker compose up -d

# 2. 应用迁移
alembic upgrade head

# 3. 启动 API 服务
cd src && uvicorn order_center.api.main:app --reload --port 8000

# 4. 验证 API
curl http://localhost:8000/health
curl -X POST http://localhost:8000/api/v1/orders/register \
  -H "Content-Type: application/json" \
  -d '{"username":"test","email":"test@test.com","password":"pass123"}'

curl -X POST http://localhost:8000/api/v1/orders/place \
  -H "Content-Type: application/json" \
  -d '{"user_id":1,"items":[{"product_id":1,"quantity":2}]}'

# 5. 运行测试
pytest tests/ -v

验收标准检查清单

验收项	标准	验证方式
模型声明	4 个模型，关系正确，约束完整	inspect(engine) 检查列和约束
用户注册	唯一性校验，重名报 409	测试 test_register_user
下单	库存扣减原子，乐观锁冲突重试	测试 test_place_order_success + test_place_order_optimistic_lock
幂等	相同 idempotency_key 不重复创建	测试 test_place_order_idempotent
订单查询	分页 + 按状态筛选	测试 test_get_user_orders
迁移	Schema 全部通过 Alembic 变更	alembic upgrade head + alembic downgrade -1 通过
测试	核心用例全绿，覆盖率 > 80%	pytest tests/ --cov=order_center
事务	关键路径有事务（下单 Service 内 commit）	代码 review

可能遇到的坑及解决方法

1. 首次运行 alembic autogenerate 生成空迁移

• 原因：Base.metadata 没有包含所有模型——因为模型模块没有被导入。
• 解决：在 alembic/env.py 中显式 import order_center.models，确保所有模型类被注册到 Base.metadata。

2. FastAPI TestClient 的 session 与测试 fixture 不一致

• 现象：TestClient 的请求使用了生产 engine 的 session，而非测试事务回滚的连接。
• 解决：通过 app.dependency_overrides 替换 get_db，使其返回测试 session。或在集成测试中使用真实的 get_db 但连接测试库。

3. version_id_col 在测试事务回滚后版本号"跳跃"

• 现象：第一个测试插入 Product(version_id=1)，事务回滚后第二个测试插入的 Product 的 version_id 从 2 开始。
• 根因：PostgreSQL 的序列在事务回滚后不重置。
• 解决：测试中不依赖具体的 version_id 值，只校验"version_id 是否增加了"。

4. FastAPI Depends 注入的 session commit 与 Service 层的 commit 冲突

• 现象：同一个请求中 session 被 commit 了两次。
• 解决：选择单一事务管理点。要么在 deps.get_db 中统一 commit/rollback，要么在 Service 中管理事务。不要两边都管。本实战采用 Service 层管理事务，deps.get_db 的 commit 作为兜底保障。

四、项目总结

交付物清单

交付物	说明
src/order_center/models/	User / Product / Order / OrderItem 四个模型，含关系映射和乐观锁
src/order_center/repository/	UserRepository / ProductRepository / OrderRepository 数据访问封装
src/order_center/service/	OrderService（下单+幂等+重试）/ ProductService
src/order_center/api/	FastAPI 路由：注册 / 下单 / 订单列表 / 商品搜索
alembic/	迁移配置 + 初始迁移 + 增量迁移示例
tests/	Repository 单元测试 / Service 集成测试 / API 端到端测试
docker-compose.yml	PostgreSQL 16 本地开发环境
pyproject.toml	依赖声明（sqlalchemy, fastapi, uvicorn, alembic, pytest）

优点与缺点

对比维度	裸 SQL 字符串 PHP 服务	SQLAlchemy MVP
SQL 注入防护	手动拼接，漏洞多	所有查询参数化，无注入风险
代码可维护性	3000 行 SQL 散落各处	Repository/Service/API 分层，责任清晰
Schema 变更	手写 ALTER TABLE，无记录	Alembic 迁移，有版本历史，可回滚
并发安全	无乐观锁，双十一超卖	version_id 乐观并发 + 重试机制
可测试性	需要真实 DB 才能测	事务回滚 + factory，秒级跑完
开发体验	改 SQL 要改多处	IDE 自动补全，类型提示，重构安全

适用场景

1. 中小型电商/交易系统的订单核心模块。
2. 需要快速验证的 MVP 项目（2 周内上线核心链路）。
3. 从遗留系统向 SQLAlchemy 迁移的第一个里程碑。
4. 作为团队 SQLAlchemy 培训的实战样板。

不适用场景：

1. 超高并发（百万 QPS）秒杀场景——需要在 MVP 基础上增加 Redis 排队、消息队列削峰等中间件。
2. 需要 CQRS/Event Sourcing 的复杂领域——MVP 是传统的 CRUD 模式。

注意事项

1. Alembic 迁移是生产部署的唯一 Schema 变更途径——绝对不要在有数据的生产库上使用 create_all。
2. 幂等键的唯一约束必须加——幂等是"最终一致性"的关键，数据库级的唯一约束是最后的防线。
3. Service 层的 commit 和 FastAPI Depends 的 commit 不要重复——选一个地方管理事务，避免 double commit。
4. 测试中不要 commit——用 flush 验证行为，依赖事务回滚保证测试隔离。

常见踩坑经验

案例 1：FastAPI 的 TestClient 绕过依赖注入

• 现象：API 测试中 Depends(get_db) 返回了生产连接的 Session 而非测试连接。
• 根因：TestClient 默认使用真实的 FastAPI app 实例，不会自动覆盖依赖。
• 修复：使用 app.dependency_overrides 注入测试 session，或连接测试库而非生产库。

案例 2：Alembic 多环境迁移混乱

• 现象：开发环境 alembic upgrade head 成功但生产环境失败。
• 根因：Alembic 的 alembic_version 表在不同环境之间不同步，或者有人直接在生产数据库上手动改了 Schema。
• 修复：建立严格的变更流程——所有 DDL 变更必须通过 Alembic。季度对生产库做 Schema 一致性巡检。

案例 3：Service 层中的重试逻辑导致数据不一致

• 现象：重试期间外部 API 回调已完成，但重试成功后数据不一致。
• 根因：重试只管理了数据库操作，没有考虑外部服务的最终状态。
• 修复：将外部服务调用放在事务之外（先 commit 数据库事务，再调外部 API）。配合幂等键确保外部调用的重试安全。

思考题

1. MVP 中的下单接口使用了乐观并发锁（version_id_col）来保护库存扣减。在 618 大促场景下（某商品 100 件库存，1 万人同时抢购），乐观锁会带来大量重试失败。请设计两种可以降低重试率的优化方案（如 Redis 预扣库存、数据库悲观锁 SELECT FOR UPDATE），并分析各自的优缺点。

2. 本章的 MVP 使用了同步的 FastAPI 框架。如果将其改为异步（asyncio + AsyncSession），需要修改哪些层次？请用代码示例指出 def get_db() → async def get_db()、Session → AsyncSession、以及 Repository 中 session.execute() 的语法差异。

参考答案参见附录 E。

延伸阅读与资源

NumPy 从入门到生产落地：全链路实战指南（科学计算/向量化）
Redis 8 实战精讲：从 CRUD 到源码，构建高可用缓存系统
Redis 实战修炼与原理进阶
Python 3实战精进：从脚本到高并发订单引擎
python入门：Rquests从菜鸟脚本到企业级SDK的网络实战圣经
Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地
MongoDB 实战进阶与内核修炼
后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战
10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析