3. 数据库模型设计与ORM基础：使用SQLAlchemy定义User、Product等核心模型，数据库迁移与初始化

在契约测试驱动的工程中，数据库模型是数据持久化的基石，也是API响应格式校验的最终数据源。本章将使用SQLAlchemy ORM定义User、Product等核心业务模型，并通过Alembic实现数据库迁移与初始化，确保模型定义与数据库Schema始终保持一致。

3.1 核心模型设计原则

模型设计需遵循以下原则，以支撑后续的契约测试与缓存校验：

  • 单一职责：每个模型对应一个业务实体，字段定义清晰，避免冗余。
  • 可追溯性：模型字段与API契约中的响应字段一一映射，便于校验。
  • 可迁移性：使用Alembic管理Schema变更，确保开发、测试、生产环境一致。
  • 性能友好：合理设置索引、关系加载策略，避免N+1查询。
设计提示：在契约测试中，模型字段的默认值、非空约束、唯一约束直接影响API响应格式的校验规则。建议在模型定义阶段就与API契约文档对齐。

3.2 使用SQLAlchemy定义核心模型

以下为User和Product模型的完整定义，包含字段、关系、约束及索引。

3.2.1 User模型

from sqlalchemy import Column, Integer, String, DateTime, Boolean, Enum, Index
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.database import Base
import enum

class UserRole(str, enum.Enum):
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(120), unique=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    role = Column(Enum(UserRole), default=UserRole.USER, nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    # 关系：一个用户可以有多个订单
    orders = relationship("Order", back_populates="user", lazy="selectin")

    # 复合索引：加速按角色和活跃状态的查询
    __table_args__ = (
        Index("ix_users_role_active", "role", "is_active"),
    )

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

3.2.2 Product模型

from sqlalchemy import Column, Integer, String, Float, Boolean, Text, DateTime, Numeric
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.database import Base

class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True, index=True)
    sku = Column(String(50), unique=True, nullable=False, index=True)
    name = Column(String(200), nullable=False)
    description = Column(Text, nullable=True)
    price = Column(Numeric(10, 2), nullable=False)
    stock_quantity = Column(Integer, default=0, nullable=False)
    is_available = Column(Boolean, default=True, nullable=False)
    category = Column(String(100), nullable=True, index=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    # 关系：一个产品可以出现在多个订单项中
    order_items = relationship("OrderItem", back_populates="product", lazy="selectin")

    def __repr__(self):
        return f"<Product(id={self.id}, sku='{self.sku}', name='{self.name}', price={self.price})>"

3.2.3 模型字段对比

模型 关键字段 约束 索引 关系
User username, email, hashed_password, role username/email唯一，role非空 主键、username、复合索引(role, is_active) 一对多 → Order
Product sku, name, price, stock_quantity, category sku唯一，price非空，stock_quantity默认0 主键、sku、category 一对多 → OrderItem

3.3 数据库迁移与初始化

使用Alembic管理数据库Schema的版本演进，确保模型变更可追溯、可回滚。

3.3.1 初始化Alembic环境

# 安装依赖
pip install alembic

# 初始化Alembic
alembic init alembic

# 配置 alembic.ini 中的数据库连接字符串
# sqlalchemy.url = postgresql://user:password@localhost:5432/contract_test_db

3.3.2 配置env.py加载模型元数据

# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# 导入所有模型，确保Base.metadata包含所有表
from app.database import Base
from app.models import User, Product, Order, OrderItem  # 导入所有模型

config = context.config
fileConfig(config.config_file_name)
target_metadata = Base.metadata

def run_migrations_offline():
    url = config.get_main_option("sqlalchemy.url")
    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
    with context.begin_transaction():
        context.run_migrations()

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)
        with context.begin_transaction():
            context.run_migrations()

3.3.3 生成并执行迁移

# 生成迁移脚本（自动检测模型变更）
alembic revision --autogenerate -m "add_user_product_models"

# 执行迁移到最新版本
alembic upgrade head

# 查看迁移历史
alembic history
迁移注意事项：
1. 生产环境执行迁移前，务必在预发布环境验证。
2. 避免在迁移脚本中直接修改已发布的数据，应使用数据迁移脚本。
3. 每次模型变更后，及时生成新的迁移版本，并提交到版本控制。

3.4 模型关系与数据完整性

在契约测试中，模型关系直接影响API响应中的嵌套数据结构。以下为Order和OrderItem模型的关系定义：

# app/models/order.py
from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey, Enum
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.database import Base
import enum

class OrderStatus(str, enum.Enum):
    PENDING = "pending"
    PAID = "paid"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"

class Order(Base):
    __tablename__ = "orders"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
    status = Column(Enum(OrderStatus), default=OrderStatus.PENDING, nullable=False)
    total_amount = Column(Float, default=0.0, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

    user = relationship("User", back_populates="orders")
    items = relationship("OrderItem", back_populates="order", lazy="selectin",
                         cascade="all, delete-orphan")

class OrderItem(Base):
    __tablename__ = "order_items"

    id = Column(Integer, primary_key=True, index=True)
    order_id = Column(Integer, ForeignKey("orders.id"), nullable=False, index=True)
    product_id = Column(Integer, ForeignKey("products.id"), nullable=False)
    quantity = Column(Integer, nullable=False, default=1)
    unit_price = Column(Float, nullable=False)

    order = relationship("Order", back_populates="items")
    product = relationship("Product", back_populates="order_items")

3.5 数据库初始化脚本

为了方便测试环境快速重建数据库，提供以下初始化脚本：

# scripts/init_db.py
from app.database import engine, Base
from app.models import User, Product, Order, OrderItem

def init_database():
    """创建所有表（仅用于开发/测试环境）"""
    Base.metadata.create_all(bind=engine)
    print("✅ 数据库表创建完成")

def drop_database():
    """删除所有表（谨慎使用）"""
    Base.metadata.drop_all(bind=engine)
    print("⚠️ 数据库表已删除")

if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "--drop":
        drop_database()
    else:
        init_database()
工程实践：在CI/CD流水线中，每次运行契约测试前执行alembic upgrade head，确保测试数据库与最新模型定义一致。测试结束后，使用事务回滚或数据库快照恢复初始状态。

3.6 模型与契约校验的映射关系

以下SVG图展示了模型字段如何映射到API响应契约中的字段，以及迁移工具如何保证Schema一致性：

模型-契约-迁移映射关系图 模型定义 → 契约校验 → 数据库迁移 映射关系 SQLAlchemy 模型 User, Product, Order 字段类型、约束、索引 关系定义 (FK, back_populates) API 响应契约 字段名、类型、格式 嵌套结构、可选/必填 校验规则 (pydantic) Alembic 迁移 自动生成迁移脚本 版本控制 (upgrade/downgrade) Schema 一致性保障 字段映射 Schema 同步 数据流说明 模型定义 → 生成迁移脚本 → 更新数据库Schema → API响应格式校验 契约测试通过后，模型变更需同步更新契约文件 模型定义 Base.metadata 迁移脚本 versions/xxx.py 数据库表 实际Schema

3.7 模型测试与验证

为确保模型定义的正确性，建议编写以下单元测试：

# tests/test_models.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.models import User, Product, Order, OrderItem

@pytest.fixture(scope="function")
def db_session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()

def test_create_user(db_session):
    user = User(username="testuser", email="test@example.com", hashed_password="hashed123")
    db_session.add(user)
    db_session.commit()
    assert user.id is not None
    assert user.role == UserRole.USER

def test_create_product_with_relationship(db_session):
    user = User(username="buyer", email="buyer@example.com", hashed_password="hashed456")
    db_session.add(user)
    db_session.commit()
    
    product = Product(sku="SKU-001", name="Test Product", price=99.99, stock_quantity=10)
    db_session.add(product)
    db_session.commit()
    
    order = Order(user_id=user.id, total_amount=99.99)
    db_session.add(order)
    db_session.commit()
    
    item = OrderItem(order_id=order.id, product_id=product.id, quantity=1, unit_price=99.99)
    db_session.add(item)
    db_session.commit()
    
    assert len(order.items) == 1
    assert order.items[0].product.name == "Test Product"
测试策略：使用SQLite内存数据库进行模型单元测试，速度快且无需外部依赖。集成测试时再连接真实数据库（如PostgreSQL），验证迁移脚本的正确性。

3.8 本章小结

通过本章学习，你已掌握：

  • 使用SQLAlchemy定义User、Product等核心业务模型，包括字段约束、索引和关系。
  • 通过Alembic实现数据库迁移的初始化、生成和执行，确保Schema版本可控。
  • 模型字段与API响应契约的映射关系，为后续契约测试提供数据基础。
  • 编写模型单元测试，验证数据完整性和关系正确性。
← 上一章 📖 返回目录 下一章 →