多对多关系

典型场景：文章和标签，一篇文章有多个标签，一个标签也可以属于多篇文章。

一、场景说明

文章表 (articles)          中间表 (article_tags)         标签表 (tags)
┌────┬──────────┐         ┌────────────┬────────┐       ┌────┬──────────┐
│ id │  title   │         │ article_id │ tag_id │       │ id │   name   │
├────┼──────────┤         ├────────────┼────────┤       ├────┼──────────┤
│  1 │ Python文 │ ── 1:N ─│     1      │   1    │─ N:1 ─│  1 │  Python  │
│  2 │ FastAPI文│         │     1      │   2    │       │  2 │  FastAPI │
└────┴──────────┘         │     2      │   2    │       │  3 │   Web    │
                          │     2      │   3    │       └────┴──────────┘
                          └────────────┴────────┘

文章 1 有标签 Python 和 FastAPI
文章 2 有标签 FastAPI 和 Web
标签 FastAPI 被文章 1 和文章 2 共用

为什么需要中间表？

一对多只需要在「多」的那张表加外键就行。但多对多两边都是「多」，一张表放不下两个外键的对应关系，所以需要一张中间表来记录「谁和谁关联」。

二、定义模型

from sqlalchemy import Column, Integer, String, ForeignKey, Table, Text, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from database import Base


# ---------- 中间表 ----------
# 用 Table 定义，不需要写 Model 类
# 两个外键组成联合主键，保证同一篇文章不会重复关联同一个标签
article_tags = Table(
    "article_tags",
    Base.metadata,
    Column("article_id", Integer, ForeignKey("articles.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True),
)


# ---------- 文章表 ----------
class Article(Base):
    __tablename__ = "articles"

    id         = Column(Integer, primary_key=True)
    title      = Column(String(200), nullable=False)
    content    = Column(Text)
    created_at = Column(DateTime, default=datetime.now)

    # secondary=article_tags — 告诉 SQLAlchemy 通过哪张中间表关联
    tags = relationship("Tag", secondary=article_tags, back_populates="articles")


# ---------- 标签表 ----------
class Tag(Base):
    __tablename__ = "tags"

    id   = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)

    articles = relationship("Article", secondary=article_tags, back_populates="tags")

三、逐行解释

1. 中间表

article_tags = Table(
    "article_tags",                  # 中间表名
    Base.metadata,                   # 注册到 Base 的 metadata 中
    Column("article_id", Integer, ForeignKey("articles.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True),
)

Table — 用来定义一张纯数据库表，不需要写 Python 类
两个 Column 都是外键，分别指向 articles 和 tags
两个 primary_key=True 组成联合主键，保证同一篇文章不会重复关联同一个标签

联合主键的效果：

# 第一次关联：正常插入
(article_id=1, tag_id=1)  →  OK

# 第二次重复关联：报错，因为联合主键冲突
(article_id=1, tag_id=1)  →  UNIQUE constraint failed

2. secondary 参数

tags = relationship("Tag", secondary=article_tags, back_populates="articles")

secondary=article_tags — 告诉 SQLAlchemy 通过 article_tags 这张中间表关联
两端都要指同一个中间表

3. 配对关系

Article.tags   ←→  Tag.articles
      ↑                  ↑
      └── secondary ─────┘
           (都指向 article_tags)

四、完整示例

4.1 建表并插入数据

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Table, Text, DateTime, select
from sqlalchemy.orm import sessionmaker, DeclarativeBase, relationship, selectinload
from datetime import datetime


class Base(DeclarativeBase):
    pass


article_tags = Table(
    "article_tags", Base.metadata,
    Column("article_id", Integer, ForeignKey("articles.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True),
)


class Article(Base):
    __tablename__ = "articles"
    id         = Column(Integer, primary_key=True)
    title      = Column(String(200), nullable=False)
    content    = Column(Text)
    created_at = Column(DateTime, default=datetime.now)
    tags       = relationship("Tag", secondary=article_tags, back_populates="articles")


class Tag(Base):
    __tablename__ = "tags"
    id       = Column(Integer, primary_key=True)
    name     = Column(String(50), unique=True, nullable=False)
    articles = relationship("Article", secondary=article_tags, back_populates="tags")


engine = create_engine("sqlite:///./test.db", connect_args={"check_same_thread": False})
Base.metadata.create_all(bind=engine)
Session = sessionmaker(bind=engine)
db = Session()


# ---------- 创建标签 ----------
tag_python = Tag(name="Python")
tag_web    = Tag(name="Web")
tag_fast   = Tag(name="FastAPI")
db.add_all([tag_python, tag_web, tag_fast])
db.commit()
db.refresh(tag_python)
db.refresh(tag_web)
db.refresh(tag_fast)

# ---------- 创建文章并关联标签 ----------
article1 = Article(title="Python 入门", content="...")
article1.tags.extend([tag_python, tag_fast])  # 给文章1关联 Python 和 FastAPI

article2 = Article(title="Web 开发", content="...")
article2.tags.extend([tag_fast, tag_web])     # 给文章2关联 FastAPI 和 Web

db.add_all([article1, article2])
db.commit()

此时数据库中的数据：

articles 表                    tags 表
┌────┬────────────┐           ┌────┬──────────┐
│ id │   title    │           │ id │   name   │
├────┼────────────┤           ├────┼──────────┤
│  1 │ Python 入门│           │  1 │ Python   │
│  2 │ Web 开发   │           │  2 │ FastAPI  │
└────┴────────────┘           │  3 │ Web      │
                              └────┴──────────┘

article_tags 中间表
┌────────────┬────────┐
│ article_id │ tag_id │
├────────────┼────────┤
│     1      │   1    │  ← 文章1 关联 Python
│     1      │   2    │  ← 文章1 关联 FastAPI
│     2      │   2    │  ← 文章2 关联 FastAPI
│     2      │   3    │  ← 文章2 关联 Web
└────────────┴────────┘

4.2 正向查询：文章 → 标签

stmt = select(Article).options(selectinload(Article.tags))
articles = db.execute(stmt).scalars().all()

for article in articles:
    tag_names = [t.name for t in article.tags]
    print(f"{article.title}: {tag_names}")

# 输出：
# Python 入门: ['Python', 'FastAPI']
# Web 开发: ['FastAPI', 'Web']

4.3 反向查询：标签 → 文章

stmt = select(Tag).options(selectinload(Tag.articles))
tags = db.execute(stmt).scalars().all()

for tag in tags:
    article_titles = [a.title for a in tag.articles]
    print(f"{tag.name}: {article_titles}")

# 输出：
# Python: ['Python 入门']
# FastAPI: ['Python 入门', 'Web 开发']
# Web: ['Web 开发']

4.4 查包含某个标签的文章

from sqlalchemy.orm import selectinload

# 查所有带 "FastAPI" 标签的文章
stmt = (
    select(Article)
    .join(Article.tags)                   # JOIN 中间表
    .where(Tag.name == "FastAPI")         # WHERE tag.name = 'FastAPI'
    .options(selectinload(Article.tags))  # 同时加载标签信息
)
articles = db.execute(stmt).scalars().all()

for article in articles:
    print(article.title)
# 输出：
# Python 入门
# Web 开发

背后的 SQL：

SELECT articles.*
FROM articles
JOIN article_tags ON articles.id = article_tags.article_id
JOIN tags ON tags.id = article_tags.tag_id
WHERE tags.name = 'FastAPI';

五、添加和移除关联

5.1 添加标签

# 给文章添加一个已有的标签
article = db.execute(select(Article).where(Article.id == 1)).scalars().first()
article.tags.append(tag_web)   # 给文章1加上 Web 标签
db.commit()

# 创建新标签并直接关联
article.tags.append(Tag(name="SQLAlchemy"))
db.commit()

5.2 移除标签

# 移除单个标签
article = db.execute(select(Article).where(Article.id == 1)).scalars().first()
article.tags = [t for t in article.tags if t.name != "Web"]
db.commit()

# 等价的写法：
for tag in article.tags:
    if tag.name == "Web":
        article.tags.remove(tag)
        break
db.commit()

注意： 移除关联不会删除 Tag 本身，只是在中间表中删了一行记录。

5.3 清空和替换

# 清空所有标签
article.tags.clear()
db.commit()

# 替换所有标签
article.tags = [tag_python, tag_fast]
db.commit()

六、中间表加额外字段

有时候关联上需要存更多信息（比如添加时间、排序权重），中间表需要改成 Model。

6.1 什么时候需要额外字段

场景	额外字段	是否需要 Model
文章 + 标签	无	不需要，用 `Table`
文章 + 标签	添加时间、添加人	需要，用 Model
学生 + 课程	成绩、选课时间	需要，用 Model
用户 + 角色	无	不需要，用 `Table`

6.2 定义带额外字段的中间表

from sqlalchemy import Column, Integer, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from database import Base


# ---------- 中间表改成 Model ----------
class ArticleTag(Base):
    __tablename__ = "article_tags"

    # 两个外键组成联合主键
    article_id = Column(Integer, ForeignKey("articles.id"), primary_key=True)
    tag_id     = Column(Integer, ForeignKey("tags.id"), primary_key=True)

    # 额外字段
    created_at = Column(DateTime, default=datetime.now)   # 关联时间
    sort_order = Column(Integer, default=0)               # 排序权重

    # 反向引用：通过中间表对象访问两边的实体
    article = relationship("Article", back_populates="tag_links")
    tag     = relationship("Tag", back_populates="article_links")


class Article(Base):
    __tablename__ = "articles"
    id    = Column(Integer, primary_key=True)
    title = Column(String(200))

    # 注意：这里关联的是中间表 Model，不是 Tag
    tag_links = relationship("ArticleTag", back_populates="article")


class Tag(Base):
    __tablename__ = "tags"
    id   = Column(Integer, primary_key=True)
    name = Column(String(50))

    article_links = relationship("ArticleTag", back_populates="tag")

6.3 使用方式

# 创建关联（带额外信息）
link = ArticleTag(article_id=1, tag_id=2, sort_order=1, created_at=datetime.now())
db.add(link)
db.commit()

# 查询：通过中间表访问
article = db.execute(select(Article).where(Article.id == 1)).scalars().first()
for link in article.tag_links:
    print(f"标签: {link.tag.name}, 排序: {link.sort_order}, 添加时间: {link.created_at}")

# 输出：
# 标签: Python, 排序: 0, 添加时间: 2024-01-15 10:30:00
# 标签: FastAPI, 排序: 1, 添加时间: 2024-01-15 10:30:00

6.4 如果还想直接访问 tags

加上一个混合属性，从中间表中提取 tag 对象：

from sqlalchemy.orm import object_session

class Article(Base):
    __tablename__ = "articles"
    id    = Column(Integer, primary_key=True)
    title = Column(String(200))

    tag_links = relationship("ArticleTag", back_populates="article")

    @property
    def tags(self):
        """快捷访问：从 tag_links 中提取 Tag 对象"""
        return [link.tag for link in self.tag_links]

article = db.execute(select(Article).where(Article.id == 1)).scalars().first()
print(article.tags)  # [<Tag(name='Python')>, <Tag(name='FastAPI')>]

七、总结

简单场景（不需要额外字段）：
  Table("中间表") + relationship(secondary=中间表)
  → 直接操作 article.tags.append(tag)

复杂场景（需要额外字段）：
  Model 中间表 + relationship(back_populates=...)
  → 操作 article.tag_links.append(ArticleTag(...))
  → 通过 link.tag 访问实际的 Tag 对象

#多对多关系

#一、场景说明

#二、定义模型

#三、逐行解释

#1. 中间表

#2. secondary 参数

#3. 配对关系

#四、完整示例

#4.1 建表并插入数据

#4.2 正向查询：文章 → 标签

#4.3 反向查询：标签 → 文章

#4.4 查包含某个标签的文章

#五、添加和移除关联

#5.1 添加标签

#5.2 移除标签

#5.3 清空和替换

#六、中间表加额外字段

#6.1 什么时候需要额外字段

#6.2 定义带额外字段的中间表

#6.3 使用方式

#6.4 如果还想直接访问 tags

#七、总结