数据库关系设计：一对多、多对多实战

📂 所属阶段：第二阶段 — 交互与数据（核心篇）
🔗 相关章节：SQLAlchemy ORM · 评论系统与交互

1. 关系类型

1.1 三种关系

一对多（One-to-Many）：
用户 ──── 多个 ──── 文章
(User)              (Post)

多对多（Many-to-Many）：
文章 ──── 多个 ──── 标签
(Post)              (Tag)
通过中间表关联

一对一（One-to-One）：
用户 ──── 1 ──── 用户资料
(User)              (Profile)

2. 一对多关系

2.1 基础配置

# app/models.py
from app.extensions import db

class Category(db.Model):
    __tablename__ = "categories"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True, nullable=False)
    slug = db.Column(db.String(50), unique=True)
    description = db.Column(db.String(200))

    # 反向引用：获取该分类下的所有文章
    posts = db.relationship("Post", back_populates="category", lazy="dynamic")

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


class Post(db.Model):
    __tablename__ = "posts"
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text)
    category_id = db.Column(db.Integer, db.ForeignKey("categories.id"))
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    # 外键关系
    category = db.relationship("Category", back_populates="posts")


class Comment(db.Model):
    __tablename__ = "comments"
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.Text, nullable=False)
    post_id = db.Column(db.Integer, db.ForeignKey("posts.id", ondelete="CASCADE"))
    author_id = db.Column(db.Integer, db.ForeignKey("users.id"))
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    post = db.relationship("Post", back_populates="comments")
    author = db.relationship("User", back_populates="comments")

3. 一对多关系查询

3.1 获取关联对象

# 获取文章的分类
post = Post.query.get(1)
print(post.category.name)  # 自动 JOIN 查询分类

# 获取分类下的所有文章
category = Category.query.get(1)
for post in category.posts:
    print(post.title)

# 反向查询数量（不加载数据，只计数）
category = Category.query.get(1)
comment_count = category.posts.count()

# 过滤关联对象
published_posts = category.posts.filter_by(is_published=True).all()

3.2 反向引用 lazy 参数

# lazy="select"（默认）：访问时单独查询（懒加载）
post.category  # 单独 SELECT * FROM categories WHERE id = ?

# lazy="joined"：查询 Post 时自动 JOIN（预加载）
post.category  # 直接从 JOIN 结果获取，无需额外查询

# lazy="dynamic"：返回查询对象而非列表
category.posts.filter(...).first()  # 可继续链式调用

# lazy="selectin"：批量预加载（推荐 N+1 问题）
posts = Post.query.options(joinedload(Post.category)).all()

3.3 解决 N+1 问题

# ❌ N+1 问题：N 篇文章会触发 N+1 次查询
posts = Post.query.all()
for post in posts:
    print(post.category.name)  # 每次访问 category 都单独查询！

# ✅ 预加载（joinedload）
from sqlalchemy.orm import joinedload
posts = Post.query.options(joinedload(Post.category)).all()
for post in posts:
    print(post.category.name)  # 只触发 1 次 JOIN 查询

# ✅ selectinload（推荐大量数据）
from sqlalchemy.orm import selectinload
posts = Post.query.options(selectinload(Post.category)).all()

# ✅ 子查询预加载
from sqlalchemy.orm import subqueryload
posts = Post.query.options(subqueryload(Post.comments)).all()

4. 多对多关系

4.1 定义中间表

# 文章与标签的多对多关系
post_tags = db.Table(
    "post_tags",
    db.Column("post_id", db.Integer, db.ForeignKey("posts.id", ondelete="CASCADE"), primary_key=True),
    db.Column("tag_id", db.Integer, db.ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True),
    db.Column("created_at", db.DateTime, default=datetime.utcnow),
)


class Tag(db.Model):
    __tablename__ = "tags"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(30), unique=True, nullable=False)
    slug = db.Column(db.String(30), unique=True)
    posts = db.relationship("Post", secondary=post_tags, back_populates="tags")


class Post(db.Model):
    __tablename__ = "posts"
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    # ... 其他字段

    tags = db.relationship("Tag", secondary=post_tags, back_populates="posts")

4.2 多对多操作

# 添加标签
post = Post.query.get(1)
tag1 = Tag.query.filter_by(name="Python").first()
tag2 = Tag.query.filter_by(name="Flask").first()
post.tags.append(tag1)
post.tags.append(tag2)
db.session.commit()

# 移除标签
post.tags.remove(tag1)

# 清空所有标签
post.tags = []

# 查询某标签下的所有文章
tag = Tag.query.filter_by(name="Python").first()
for post in tag.posts:
    print(post.title)

# 查询同时拥有多个标签的文章
from sqlalchemy import and_
posts = Post.query.join(Post.tags).filter(
    Tag.name.in_(["Python", "Flask"])
).group_by(Post.id).having(
    func.count(Tag.id) == 2
).all()

5. 自引用关系（树形结构）

5.1 评论树（自引用一对多）

class Comment(db.Model):
    __tablename__ = "comments"
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.Text, nullable=False)
    post_id = db.Column(db.Integer, db.ForeignKey("posts.id"))
    parent_id = db.Column(db.Integer, db.ForeignKey("comments.id"), nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    # 自引用：父评论
    replies = db.relationship(
        "Comment",
        backref=db.backref("parent", remote_side=[id]),
        lazy="dynamic",
        cascade="all, delete-orphan"
    )


# 查询
comment = Comment.query.get(1)
parent = comment.parent        # 上级评论
replies = comment.replies.all()  # 所有回复

6. cascade 配置

# cascade：控制级联操作

db.relationship("Post",
    back_populates="author",
    cascade="all, delete-orphan"
)
cascade 值说明
all所有级联操作
save-update添加父对象时自动添加子对象
delete删除父对象时删除子对象
delete-orphan删除父对象时orphan（孤儿子对象）也删除
merge合并操作

7. 小结

# 关系速查

# 一对多
author_id = Column(ForeignKey("users.id"))
author = relationship("User", back_populates="posts")
posts = relationship("Post", back_populates="author")

# 多对多
中间表 = Table("中间表", Column(...), Column(...))
tags = relationship("Tag", secondary=中间表, back_populates="posts")

# 自引用
parent_id = Column(ForeignKey("comments.id"))
parent = relationship("Comment", backref=backref("replies", remote_side=[id]))

# 预加载
joinedload(Model.field)        # JOIN 预加载
selectinload(Model.field)      # 批量预加载
subqueryload(Model.field)      # 子查询预加载

💡 最佳实践：使用 lazy="dynamic" 处理可能很大的关联集合，用 selectinload 或 joinedload 解决 N+1 查询问题。

🔗 扩展阅读