一、项目背景

“为什么商品标签查不出来？我明明给商品加了’热门’和’新品’两个标签，但后台筛选’热门标签’时只出来了一部分商品？”

星云电商运营团队在促销活动前发现了这个数据异常。开发排查后发现，问题出在商品-标签的多对多关系设计上。当时为了赶工期，开发直接在 products 表中加了一个 VARCHAR 字段 tags，用逗号分隔存储标签（如 "热门,新品,有机"）。这种设计虽然简单，但带来了三个致命问题：

1. 查询性能极差：查找所有"热门"标签的商品，只能用 WHERE tags LIKE '%热门%'——全表扫描，且 LIKE '%xx%' 无法走索引。随着商品数突破 50 万，这个查询每次耗时超过 3 秒。

2. 数据完整性问题：一个标签名改了（“有机食品"→"天然有机”），需要更新所有包含该标签的商品记录——一个 UPDATE 扫全表，而且容易遗漏。

3. 并发修改冲突：两个运营同时对同一商品修改标签，后保存的覆盖了先保存的（因为整个 tags 字段被整行替换），导致标签丢失。

另一个场景是购物车——用户可以将商品加入购物车，这个关系需要额外记录"加入数量"和"加入时间"。它不是单纯的多对多，而是带额外字段的关联关系（“用户-商品-数量-加购时间”）。

本章将覆盖两种核心模式：纯多对多（商品↔标签，通过二级表连接）和关联对象模式（用户-购物车-商品，带额外字段的中间模型），并通过实战对比两者的适用场景。

二、项目设计

场景：周一上午，小胖抱着薯片，小白端着茶，大师开讲。

小胖：“大师，我们商品上万条，标签管理靠的是那个烂熟了的逗号分隔字符串。您给说说，正常的多对多怎么做？”

大师：“多对多，用麻将来打比方最合适。一张产品表、一张标签表，中间要有一张’关联表’——就像你坐麻将桌，四个人之间通过一副牌连接。这副牌就是二级表。”

小白：“我懂了，一张 product_tags 表，两列：product_id 和 tag_id。”

大师：“没错。最简单的多对多就是一张纯连接表——只包含两个外键，没有任何额外列。SQLAlchemy 中用 Table 对象声明它，“不用做成 ORM 模型”——然后两边用 relationship(secondary=...) 引用它。”

# 二级表：纯连接，不映射为模型
product_tag_assoc = Table(
    "product_tags", Base.metadata,
    Column("product_id", ForeignKey("products.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)

# Product 模型
class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(200))
    tags: Mapped[list["Tag"]] = relationship(secondary=product_tag_assoc, back_populates="products")

# Tag 模型
class Tag(Base):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    products: Mapped[list["Product"]] = relationship(secondary=product_tag_assoc, back_populates="tags")

小胖：“技术映射：二级表 + secondary = 多对多的桥梁。所以 product.tags.append(tag) 会自动在中间表里插一条记录？”

大师：“完全正确。当你在 Python 层操作 product.tags.append(hot_tag)，SQLAlchemy 在 flush 时自动生成 INSERT INTO product_tags VALUES (product_id, tag_id)。删除也是如此——从列表中 remove 就会自动删除中间表记录。”

小白：“那购物车呢？用户把商品加入购物车，不只是’用户和商品的关联’——还有加购数量和加购时间。如果中间表有额外字段，还能用 secondary 吗？”

大师：“不能。secondary 适用于纯连接表——中间表除了外键没有其他字段。如果中间表有额外字段（如数量、时间、价格），就需要用关联对象模式——将中间表也声明为完整的 ORM 模型。”

# 关联对象模式：中间表是一个完整的 ORM 模型
class CartItem(Base):
    __tablename__ = "cart_items"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"))
    quantity: Mapped[int] = mapped_column(Integer, default=1)      # 额外字段
    added_at: Mapped[datetime] = mapped_column(DateTime, default=func.now())  # 额外字段

    # 两端的关联
    user: Mapped["User"] = relationship(back_populates="cart_items")
    product: Mapped["Product"] = relationship(back_populates="cart_items")

小胖：“那 User 和 Product 两边怎么声明？”

大师：

class User(Base):
    cart_items: Mapped[list["CartItem"]] = relationship(back_populates="user")

class Product(Base):
    cart_items: Mapped[list["CartItem"]] = relationship(back_populates="product")

小白：“技术映射：关联对象 = 中间表升格为完整模型，承载额外业务字段。这两种模式怎么选？”

大师：“原则很简单——中间表有没有自己的业务字段？”

特征	选 secondary	选关联对象
中间表仅有外键	✓	也可以
中间表有额外字段	✗	✓
需要独立查询中间表	✗	✓
只需要简单增删关联	✓	略显啰嗦

小胖：“技术映射：secondary = 红娘牵线（只有牵手关系）；关联对象 = 婚姻登记（有结婚日期、财产协议）。”

小白：“最后一个问题：多对多查询时，如果我只想查’带有热门标签’且’库存大于 0’的商品，怎么高效地写？”

大师：“用 any() 或 has() 进行关联子查询：”

# 查询拥有'热门'标签的商品
stmt = select(Product).where(
    Product.tags.any(Tag.name == "热门"),
    Product.inventory > 0,
)

大师：“底层生成的 SQL 是 WHERE EXISTS (SELECT 1 FROM product_tags JOIN tags ON ... WHERE tags.name = '热门')，比 LIKE '%热门%' 高效得多。”

三、项目实战

实战目标

实现两个完整功能：商品-标签多对多（secondary 模式），以及购物车（关联对象模式，带加购数量和加购时间）。

步骤一：模型声明

"""ch11_many_to_many.py —— 多对多与关联对象实战"""

from sqlalchemy import (
    create_engine, String, Integer, Numeric, DateTime, Boolean,
    ForeignKey, Table, Column, UniqueConstraint, Index,
    text, func, select, and_, or_, exists,
)
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship,
    Session, sessionmaker,
)
from datetime import datetime
from typing import Optional, List

engine = create_engine(
    "postgresql+psycopg://nebula:nebula_dev@localhost:5432/order_center",
    echo=True,
)

class Base(DeclarativeBase):
    pass

# =============================================
# 模式一：secondary 多对多——商品↔标签
# =============================================

# 纯连接表（不映射为模型，只有两个外键）
product_tag_assoc = Table(
    "product_tag_assoc", Base.metadata,
    Column("product_id", ForeignKey("m2m_products.id", ondelete="CASCADE"), primary_key=True),
    Column("tag_id", ForeignKey("m2m_tags.id", ondelete="CASCADE"), primary_key=True),
    # 可以加一个联合索引提升反向查询
    Index("ix_assoc_tag_product", "tag_id", "product_id"),
)

class Tag(Base):
    __tablename__ = "m2m_tags"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, comment="标签名")
    description: Mapped[Optional[str]] = mapped_column(String(200))
    is_active: Mapped[bool] = mapped_column(Boolean, server_default=text("TRUE"))
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

    # 多对多：一个标签可以属于多个商品
    products: Mapped[List["Product"]] = relationship(
        secondary=product_tag_assoc, back_populates="tags"
    )

    def __repr__(self):
        return f"<Tag({self.name})>"

class Product(Base):
    __tablename__ = "m2m_products"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    sku: Mapped[str] = mapped_column(String(30), unique=True, nullable=False)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    unit_price: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False)
    inventory: Mapped[int] = mapped_column(Integer, server_default=text("0"))
    status: Mapped[str] = mapped_column(String(20), server_default=text("'online'"))
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

    # 多对多：一个商品可以有多个标签
    tags: Mapped[List["Tag"]] = relationship(
        secondary=product_tag_assoc, back_populates="products"
    )

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

# =============================================
# 模式二：关联对象——用户↔购物车↔商品
# =============================================

class User(Base):
    __tablename__ = "m2m_users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

    # 一对多：一个用户有多条购物车明细
    cart_items: Mapped[List["CartItem"]] = relationship(
        back_populates="user", cascade="all, delete-orphan"
    )

class CartItem(Base):
    """关联对象——购物车明细（中间表升格为完整模型）"""
    __tablename__ = "m2m_cart_items"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)

    # 关联外键
    user_id: Mapped[int] = mapped_column(ForeignKey("m2m_users.id", ondelete="CASCADE"), nullable=False)
    product_id: Mapped[int] = mapped_column(ForeignKey("m2m_products.id", ondelete="CASCADE"), nullable=False)

    # === 额外业务字段 ===
    quantity: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("1"))
    added_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    selected: Mapped[bool] = mapped_column(Boolean, server_default=text("TRUE"), comment="是否勾选")

    # 双端关联
    user: Mapped["User"] = relationship(back_populates="cart_items")
    product: Mapped["Product"] = relationship()  # 注意：Product 不持有 cart_items 的反向引用

    # 唯一约束：同一用户对同一商品只能有一条购物车记录
    __table_args__ = (
        UniqueConstraint("user_id", "product_id", name="uq_cart_user_product"),
    )

    def __repr__(self):
        return f"<CartItem({self.product_id} x{self.quantity})>"

Base.metadata.create_all(engine)
SessionFactory = sessionmaker(bind=engine, autocommit=False, autoflush=False)

步骤二：商品-标签多对多操作

# =============================================
# 多对多实战：商品 ↔ 标签
# =============================================

print("=== 多对多：商品 ↔ 标签 ===")

with SessionFactory() as session:
    # 1. 创建标签
    tags_data = ["热门", "新品", "有机", "限时特惠", "进口"]
    tag_objs = {}
    for name in tags_data:
        tag = Tag(name=name)
        tag_objs[name] = tag
        session.add(tag)
    session.flush()
    print(f"创建了 {len(tag_objs)} 个标签")

    # 2. 创建商品并添加标签
    products_data = [
        {"sku": "P-001", "title": "有机全麦面包", "price": 15.80, "tags": ["热门", "有机"]},
        {"sku": "P-002", "title": "蓝牙耳机 Pro", "price": 299.00, "tags": ["热门", "新品", "限时特惠"]},
        {"sku": "P-003", "title": "进口咖啡豆", "price": 89.00, "tags": ["进口", "新品"]},
        {"sku": "P-004", "title": "碳钢不粘锅", "price": 299.00, "tags": ["热门"]},
    ]
    for pd in products_data:
        product = Product(sku=pd["sku"], title=pd["title"], unit_price=pd["price"])
        for tag_name in pd["tags"]:
            product.tags.append(tag_objs[tag_name])  # 通过 secondary 自动写入中间表
        session.add(product)
    session.commit()
    print(f"创建了 {len(products_data)} 个商品并关联标签")

    # 3. 查询：带"热门"标签且库存>0的商品
    print("\n--- 查询：热门标签 + 有库存 ---")
    stmt = (
        select(Product)
        .where(
            Product.tags.any(Tag.name == "热门"),
            Product.inventory >= 0,
        )
        .order_by(Product.unit_price)
    )
    for product in session.execute(stmt).scalars():
        tag_names = [t.name for t in product.tags]
        print(f"  [{product.sku}] {product.title} ¥{product.unit_price} 标签: {tag_names}")

    # 4. 查询：某个标签下的商品数量统计
    print("\n--- 统计：每个标签的商品数 ---")
    stmt = (
        select(Tag.name, func.count(Product.id))
        .select_from(Tag)
        .join(product_tag_assoc, Tag.id == product_tag_assoc.c.tag_id)
        .join(Product, Product.id == product_tag_assoc.c.product_id)
        .group_by(Tag.name)
        .order_by(func.count(Product.id).desc())
    )
    for row in session.execute(stmt):
        print(f"  {row[0]}: {row[1]} 个商品")

    # 5. 验证：从标签端删除关联
    print("\n--- 删除标签关联 ---")
    hot_tag = tag_objs["热门"]
    product_p2 = session.execute(select(Product).where(Product.sku == "P-002")).scalars().one()
    product_p2.tags.remove(hot_tag)  # 从 Python 列表移除
    session.flush()  # 自动 DELETE FROM product_tag_assoc
    remaining = [t.name for t in product_p2.tags]
    print(f"  P-002 移除'热门'标签后: {remaining}")
    session.rollback()  # 仅演示

session.close()
print()

运行结果：

=== 多对多：商品 ↔ 标签 ===
创建了 5 个标签
创建了 4 个商品并关联标签

--- 查询：热门标签 + 有库存 ---
  [P-001] 有机全麦面包 ¥15.80 标签: ['热门', '有机']
  [P-004] 碳钢不粘锅 ¥299.00 标签: ['热门']
  [P-002] 蓝牙耳机 Pro ¥299.00 标签: ['热门', '新品', '限时特惠']

--- 统计：每个标签的商品数 ---
  热门: 3 个商品
  新品: 2 个商品
  有机: 1 个商品
  限时特惠: 1 个商品
  进口: 1 个商品

步骤三：购物车——关联对象模式

# =============================================
# 关联对象实战：用户 → 购物车 → 商品
# =============================================

print("\n=== 关联对象：用户 ↔ 购物车 ===")

with SessionFactory() as session:
    # 1. 创建用户
    user = User(username="小胖")
    session.add(user)
    session.flush()

    # 2. 获取商品
    p1 = session.execute(select(Product).where(Product.sku == "P-001")).scalars().one()
    p2 = session.execute(select(Product).where(Product.sku == "P-002")).scalars().one()

    # 3. 添加商品到购物车
    ci1 = CartItem(user=user, product=p1, quantity=3)
    ci2 = CartItem(user=user, product=p2, quantity=1, selected=False)
    session.add_all([ci1, ci2])
    session.flush()

    print(f"小胖的购物车（共 {len(user.cart_items)} 件）:")
    for ci in user.cart_items:
        print(f"  {ci.product.title} x{ci.quantity} (勾选:{ci.selected}) 加入:{ci.added_at}")

    # 4. 修改购物车数量（直接修改关联对象字段）
    ci1.quantity = 5
    session.flush()
    print(f"\n修改数量后: {p1.title} x{ci1.quantity}")

    # 5. 查询：统计每个用户的购物车商品总价
    print("\n--- 购物车总价统计 ---")
    stmt = (
        select(
            User.username,
            func.sum(CartItem.quantity * Product.unit_price).label("total"),
            func.count(CartItem.id).label("item_count"),
        )
        .select_from(User)
        .join(CartItem, User.id == CartItem.user_id)
        .join(Product, Product.id == CartItem.product_id)
        .where(CartItem.selected == True)
        .group_by(User.username)
    )
    for row in session.execute(stmt):
        print(f"  {row.username}: {row.item_count} 件商品, 总计 ¥{row.total:.2f}")

    # 6. 唯一约束验证：同一用户不能重复添加同一商品
    print("\n--- 唯一约束验证 ---")
    try:
        dup = CartItem(user=user, product=p1, quantity=2)
        session.add(dup)
        session.flush()
        print("  错误：不应该添加成功！")
    except Exception as e:
        print(f"  符合预期：添加失败（{type(e).__name__}）")
        session.rollback()

    # 7. 从购物车移除商品
    print("\n--- 移除购物车项 ---")
    item_to_remove = user.cart_items[1]  # p2
    session.delete(item_to_remove)
    session.flush()
    print(f"  移除 {item_to_remove.product.title} 后: 购物车剩余 {len(user.cart_items)} 件")

    session.rollback()  # 演示用，回滚

session.close()
print()

步骤四：合并查询——两种模式的对比使用

# =============================================
# 对比：两种模式下的查询差异
# =============================================

print("=== 两种模式的查询差异 ===")

with SessionFactory() as session:
    # 模式一（secondary）：查商品和标签
    print("--- secondary 模式：Product → Tags ---")
    stmt = (
        select(Product.title, func.string_agg(Tag.name, ", ").label("tag_list"))
        .select_from(Product)
        .join(product_tag_assoc, Product.id == product_tag_assoc.c.product_id, isouter=True)
        .join(Tag, Tag.id == product_tag_assoc.c.tag_id, isouter=True)
        .group_by(Product.id)
    )
    for row in session.execute(stmt):
        print(f"  {row.title}: [{row.tag_list}]")

    # 模式二（关联对象）：查购物车明细
    print("\n--- 关联对象模式：CartItem 明细 ---")
    # 这里 CartItem 本身就是模型，可以直接查询、排序、分页
    stmt = (
        select(CartItem, Product.title, Product.unit_price)
        .join(Product, CartItem.product_id == Product.id)
        .where(CartItem.selected == True)
        .order_by(CartItem.added_at.desc())
    )
    for row in session.execute(stmt):
        ci = row[0]
        print(f"  {ci.quantity} x {row.title} @ ¥{row.unit_price} (加入:{ci.added_at})")

完整代码清单

"""ch11_many_to_many_complete.py —— 多对多与关联对象完整示例"""

from sqlalchemy import (
    create_engine, String, Integer, Numeric, DateTime, Boolean,
    ForeignKey, Table, Column, UniqueConstraint, Index,
    text, func, select,
)
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship,
    sessionmaker,
)
from datetime import datetime
from typing import Optional, List
from order_center.config import DATABASE_URL

engine = create_engine(DATABASE_URL, echo=False)

class Base(DeclarativeBase):
    pass

# 二级表——纯连接
product_tag_assoc = Table(
    "pt_assoc", Base.metadata,
    Column("product_id", ForeignKey("pt_products.id", ondelete="CASCADE"), primary_key=True),
    Column("tag_id", ForeignKey("pt_tags.id", ondelete="CASCADE"), primary_key=True),
)

class Tag(Base):
    __tablename__ = "pt_tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    products: Mapped[List["Product"]] = relationship(secondary=product_tag_assoc, back_populates="tags")

class Product(Base):
    __tablename__ = "pt_products"
    id: Mapped[int] = mapped_column(primary_key=True)
    sku: Mapped[str] = mapped_column(String(30), unique=True, nullable=False)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    unit_price: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False)
    inventory: Mapped[int] = mapped_column(Integer, server_default=text("0"))
    tags: Mapped[List["Tag"]] = relationship(secondary=product_tag_assoc, back_populates="products")

class User(Base):
    __tablename__ = "pt_users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    cart_items: Mapped[List["CartItem"]] = relationship(back_populates="user", cascade="all, delete-orphan")

class CartItem(Base):
    __tablename__ = "pt_cart_items"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("pt_users.id", ondelete="CASCADE"))
    product_id: Mapped[int] = mapped_column(ForeignKey("pt_products.id", ondelete="CASCADE"))
    quantity: Mapped[int] = mapped_column(Integer, server_default=text("1"))
    selected: Mapped[bool] = mapped_column(Boolean, server_default=text("TRUE"))
    added_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    user: Mapped["User"] = relationship(back_populates="cart_items")
    product: Mapped["Product"] = relationship()
    __table_args__ = (UniqueConstraint("user_id", "product_id"),)

Base.metadata.create_all(engine)
Factory = sessionmaker(bind=engine)

def add_tags_to_product(session, product: Product, tag_names: list[str]) -> None:
    """给商品添加标签（不存在则创建）"""
    for name in tag_names:
        tag = session.execute(select(Tag).where(Tag.name == name)).scalars().first()
        if not tag:
            tag = Tag(name=name)
            session.add(tag)
        if tag not in product.tags:
            product.tags.append(tag)

def add_to_cart(session, user: User, product: Product, quantity: int = 1) -> CartItem:
    """添加商品到购物车（已存在则累加数量）"""
    ci = session.execute(
        select(CartItem).where(CartItem.user_id == user.id, CartItem.product_id == product.id)
    ).scalars().first()
    if ci:
        ci.quantity += quantity
    else:
        ci = CartItem(user=user, product=product, quantity=quantity)
        session.add(ci)
    return ci

if __name__ == "__main__":
    with Factory() as s:
        s.add_all([User(username="u1"), User(username="u2")])
        s.add(Product(sku="P1", title="T1", unit_price=10))
        s.commit()
        print("初始化完成")

可能遇到的坑及解决方法

1. secondary 表忘了在两边都声明导致关系失踪

• 现象：product.tags 有值但 tag.products 为空列表。
• 原因：只有一边写了 relationship(secondary=...)，另一边没写或写错了。
• 解决：两边都要声明 relationship(secondary=...) 并正确设置 back_populates。

2. 关联对象模式下漏配 UniqueConstraint 导致重复加购

• 现象：同一用户多次添加同一商品到购物车，创建了多条记录而不是累加数量。
• 原因：没有 UniqueConstraint("user_id", "product_id")，数据库层不阻止重复。
• 解决：加上唯一约束，并在代码层做检查（已存在则累加数量）。

3. secondary 表的 ondelete 级联顺序问题

• 现象：删除商品时中间表记录未自动删除（遗留孤行）。
• 原因：ForeignKey(ondelete="CASCADE") 只处理 in-db 级联，SQLAlchemy 的 ORM 层没有自动清理。
• 解决：配置 relationship(passive_deletes=True) 或手工维护中间表。

4. 关联对象模式的 cascade 误配导致数据丢失

• 现象：删除用户时，CartItem 被级联删除（符合预期），但 session.delete(cart_item) 时连 Product 也被删了。
• 根因：CartItem.product 的 relationship 不小心配了 cascade="all"。
• 解决：CartItem.product 不配置 cascade，只有 User.cart_items 配置 delete-orphan。

测试验证

# tests/test_ch11_m2m.py
import pytest
from sqlalchemy import create_engine, String, Integer, Numeric, ForeignKey, Table, Column, text, func, select, UniqueConstraint
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker
from datetime import datetime

@pytest.fixture
def engine():
    return create_engine("sqlite:///:memory:", echo=False)

@pytest.fixture
def models(engine):
    class Base(DeclarativeBase):
        pass

    pt_assoc = Table("pt_assoc", Base.metadata,
        Column("product_id", ForeignKey("products.id"), primary_key=True),
        Column("tag_id", ForeignKey("tags.id"), primary_key=True),
    )

    class Tag(Base):
        __tablename__ = "tags"
        id: Mapped[int] = mapped_column(primary_key=True)
        name: Mapped[str] = mapped_column(String(50), unique=True)
        products: Mapped[list["Product"]] = relationship(secondary=pt_assoc, back_populates="tags")

    class Product(Base):
        __tablename__ = "products"
        id: Mapped[int] = mapped_column(primary_key=True)
        title: Mapped[str] = mapped_column(String(200))
        tags: Mapped[list["Tag"]] = relationship(secondary=pt_assoc, back_populates="tags")

    class CartItem(Base):
        __tablename__ = "cart"
        id: Mapped[int] = mapped_column(primary_key=True)
        user_name: Mapped[str] = mapped_column(String(50))
        product_id: Mapped[int] = mapped_column(ForeignKey("products.id"))
        quantity: Mapped[int] = mapped_column(Integer, default=1)
        product: Mapped["Product"] = relationship()
        __table_args__ = (UniqueConstraint("user_name", "product_id"),)

    Base.metadata.create_all(engine)
    return Tag, Product, CartItem

def test_secondary_m2m(models, engine):
    """验证 secondary 多对多"""
    Tag, Product, _ = models
    Factory = sessionmaker(bind=engine)
    with Factory() as s:
        t = Tag(name="hot")
        p = Product(title="test")
        p.tags.append(t)
        s.add(p)
        s.commit()
        assert len(p.tags) == 1
        assert p.tags[0].name == "hot"
        assert len(t.products) == 1  # 双向同步

def test_cart_item_unique(models, engine):
    """验证关联对象唯一约束"""
    _, Product, CartItem = models
    Factory = sessionmaker(bind=engine)
    with Factory() as s:
        p = Product(title="item")
        s.add(p)
        s.flush()
        s.add(CartItem(user_name="u1", product_id=p.id, quantity=1))
        s.commit()
        # 尝试插入重复
        import sqlite3
        with pytest.raises(Exception):
            s.add(CartItem(user_name="u1", product_id=p.id, quantity=2))
            s.flush()
        s.rollback()

四、项目总结

优点与缺点

对比维度	逗号分隔字符串	secondary 多对多	关联对象模式
查询性能	LIKE 全表扫描	EXISTS 子查询走索引	直接 JOIN 走索引
数据完整性	无约束，随便写	外键约束 + 唯一约束	外键 + 唯一 + 额外字段约束
额外字段	不支持	不支持	支持任意额外字段
维护成本	标签改名需全表更新	改关联表一条记录	改关联模型一条记录
学习成本	极低	低	中

适用场景

secondary 多对多适用场景：

1. 商品↔标签、文章↔分类、用户↔角色——纯关联，无额外字段。
2. 只需要增删关联关系，不需要记录"何时关联"、"谁操作"等元信息。
3. 标签/分类等维度需要独立管理（改名、统计、迁移）。

关联对象模式适用场景：

1. 购物车——需要"数量"和"加入时间"。
2. 订单明细——需要"单价、数量、折扣"等。
3. 用户-项目权限——需要"权限生效/失效时间"。
4. 任何需要查询中间表本身的统计场景。

不推荐场景：

1. 只有两个外键的简单关联（用 secondary 即可）。
2. 数据量超大的场景（百万级关联记录）——可能需要反范式设计。

注意事项

1. secondary 表中 primary_key=True 很重要：联合主键不仅保证唯一性，还提供良好的查找性能。
2. 关联对象模式中的 relationship 要区分 cascade：不要误将关联对象的 cascade 配到非关联方（如 Product）。
3. back_populates 在 secondary 中同样有效：但要显式声明，避免只写了一边的 relationship。
4. any() 查询生成的 SQL 是 EXISTS 子查询：在大数据量下确保相关列有索引。

常见踩坑经验

案例 1：secondary 表上的 delete 行为异常

• 现象：从 product.tags 中 remove 了一个 tag，但该 tag 在 tag 表中也被删除了。
• 根因：在 Tag 的 relationship 中误写了 cascade="all, delete-orphan"。
• 修复：secondary 多对多的 relationship 不应配置 cascade，删除关联仅删除中间表记录。

案例 2：关联对象中忘记配置索引导致查询慢

• 现象：查某用户购物车，随着购物车记录增多（10 万+），查询越来越慢。
• 根因：CartItem 表的 user_id 和 product_id 没有索引。
• 修复：为外键列加上 index=True，必要时加复合索引。

案例 3：any() 与 has() 混淆导致生成错误的 SQL

• 现象：想查"拥有某标签的商品"，用了 Tag.products.has(...) 而不是 Product.tags.any(...)，SQL 语义颠倒了。
• 根因：any() 是从"多"端查关系，has() 是从"一"端查关系。两者生成的 SQL 不同。
• 修复：Product.tags.any(Tag.name == "xxx") 和 Tag.products.has(Product.inventory > 0)。

思考题

1. 假设你需要在购物车功能中添加"加购时的商品价格快照"。即：用户加购时锁定当时的商品价格，即使后续商品涨价，购物车中仍显示加入时的价格。请问这个需求应该修改哪个模型？是 Product、CartItem 还是新建一个表？为什么？

2. 你的商品表中每件商品平均有 3 个标签，系统有 100 万件商品。如果用 Product.tags.any() 来查询"同时拥有标签 A 和标签 B"的商品，生成的 SQL 是什么样的？这个 SQL 在 100 万数据量下的性能表现如何？如何优化？

---

参考答案参见附录 E。

延伸阅读与资源

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