一对多关系

最常用的关系类型:一个用户有多篇文章,一篇文章属于一个用户。

一、场景说明

用户表 (users)                 文章表 (articles)
┌────┬──────┐                 ┌────┬──────────┬─────────┬───────────┐
│ id │ name │                 │ id │  title   │ content │ author_id │
├────┼──────┤                 ├────┼──────────┼─────────┼───────────┤
│  1 │ 张三 │  ────── 1:N ──  │  1 │ 第一篇   │  ...    │     1     │
│  2 │ 李四 │                 │  2 │ 第二篇   │  ...    │     1     │
└────┴──────┘                 │  3 │ 李四的文 │  ...    │     2     │
                              └────┴──────────┴─────────┴───────────┘
  • 一个用户可以有多篇文章(一对多)
  • 一篇文章只能属于一个用户(多对一)
  • author_id 是外键,指向 users.id

二、定义模型

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


class User(Base):
    __tablename__ = "users"

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

    # relationship:Python 层面的访问快捷方式,不会在数据库中生成字段
    # "Article" — 关联的模型类名(用字符串是因为 Article 在后面才定义)
    # back_populates — 双向关联,指向对方的属性名
    articles = relationship("Article", back_populates="author")


class Article(Base):
    __tablename__ = "articles"

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

    author = relationship("User", back_populates="articles")

三、逐行解释

1. ForeignKey — 外键

author_id = Column(Integer, ForeignKey("users.id"), nullable=False)
  • ForeignKey("users.id") — 告诉数据库:author_id 的值必须是 users 表中已存在的 id
  • 这个字段真实存在于数据库中,会在 articles 表里生成 author_id
  • 如果插入一个不存在的 author_id,数据库会报错

ForeignKey 的字符串格式是 "表名.字段名"

ForeignKey("users.id")        # 正确:表名.字段名
ForeignKey("User.id")         # 错误:这里写的是数据库表名,不是 Python 类名

2. relationship — 关系映射

# User 端
articles = relationship("Article", back_populates="author")

# Article 端
author = relationship("User", back_populates="articles")

relationship 不会在数据库中生成任何字段,它只是给 Python 对象加了一个快捷属性。

有了 relationship 之后,可以这样访问:

# 没有 relationship:需要自己写查询
articles = db.execute(select(Article).where(Article.author_id == user.id)).scalars().all()

# 有了 relationship:直接用属性访问
articles = user.articles   # 自动帮你查了

3. back_populates 配对规则

User.articles    ←→  Article.author
     ↑                      ↑
     └── back_populates ────┘

两端的 back_populates 互相指向对方的属性名

# User 里定义了 articles
articles = relationship("Article", back_populates="author")
#                                       ↑ 这里写 Article 端的属性名:author

# Article 里定义了 author
author = relationship("User", back_populates="articles")
#                                    ↑ 这里写 User 端的属性名:articles

配对错了会报错或者单向不生效。

四、完整示例

4.1 建表并插入数据

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


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"
    id   = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    articles = relationship("Article", back_populates="author")


class Article(Base):
    __tablename__ = "articles"
    id         = Column(Integer, primary_key=True)
    title      = Column(String(200), nullable=False)
    content    = Column(Text)
    author_id  = Column(Integer, ForeignKey("users.id"), nullable=False)
    created_at = Column(DateTime, default=datetime.now)
    author     = relationship("User", back_populates="articles")


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


# ---------- 插入数据 ----------
# 创建两个用户
user1 = User(name="张三")
user2 = User(name="李四")
db.add_all([user1, user2])
db.commit()
db.refresh(user1)
db.refresh(user2)
print(f"张三的 id: {user1.id}")  # 1
print(f"李四的 id: {user2.id}")  # 2

# 给张三创建两篇文章
article1 = Article(title="Python 入门", content="...", author_id=user1.id)
article2 = Article(title="FastAPI 教程", content="...", author_id=user1.id)
# 给李四创建一篇文章
article3 = Article(title="SQLAlchemy 笔记", content="...", author_id=user2.id)
db.add_all([article1, article2, article3])
db.commit()

4.2 正向查询:用户 → 文章

# 查张三
user = db.execute(select(User).where(User.name == "张三")).scalars().first()

# 直接访问 user.articles,不需要写 SQL
print(user.name)        # 张三
print(user.articles)    # [<Article(title='Python 入门')>, <Article(title='FastAPI 教程')>]

# 遍历
for article in user.articles:
    print(f"  - {article.title}")
# 输出:
#   - Python 入门
#   - FastAPI 教程

背后的 SQL:

-- 第一步:查用户
SELECT * FROM users WHERE name = '张三';

-- 第二步:访问 user.articles 时自动触发
SELECT * FROM articles WHERE author_id = 1;

4.3 反向查询:文章 → 用户

# 查一篇文章
article = db.execute(select(Article).where(Article.id == 1)).scalars().first()

# 直接访问 article.author
print(article.title)        # Python 入门
print(article.author.name)  # 张三

背后的 SQL:

SELECT * FROM articles WHERE id = 1;
-- SQLAlchemy 根据 author_id 自动查对应的 User
SELECT * FROM users WHERE id = 1;

4.4 通过 relationship 添加关联

# 方式一:手动写 author_id
article = Article(title="新文章", content="...", author_id=user1.id)
db.add(article)
db.commit()

# 方式二:通过 relationship 添加(不用手动写 author_id)
user = db.execute(select(User).where(User.id == 1)).scalars().first()
user.articles.append(Article(title="又一篇新文章", content="..."))
db.commit()
# SQLAlchemy 自动把 author_id 设为 user.id

五、预加载(避免 N+1 问题)

5.1 什么是 N+1 问题

# 查出 10 个用户
users = db.execute(select(User)).scalars().all()

# 遍历时访问每个用户的文章
for user in users:
    print(user.articles)  # 每次访问都会触发一条新 SQL!

总共有 10 个用户,就会多发 10 条 SQL,加上查用户的 1 条,总共 11 条。这就是 N+1 问题。

5.2 用 joinedload 解决

from sqlalchemy.orm import joinedload

# 一条 SQL 把用户和文章都查出来(用 LEFT JOIN)
stmt = select(User).options(joinedload(User.articles))
users = db.execute(stmt).unique().scalars().all()

for user in users:
    print(user.articles)  # 不会再触发新 SQL

背后的 SQL:

SELECT users.*, articles.*
FROM users
LEFT JOIN articles ON users.id = articles.author_id;
  • joinedload — 用 JOIN 一次查完,适合关联数据量不大的情况
  • .unique() — JOIN 会产生重复行(一个用户有多篇文章就有多行),必须加 .unique() 去重

5.3 用 selectinload 解决

from sqlalchemy.orm import selectinload

# 分两条 SQL 查:先查用户,再用 IN 一次查所有文章
stmt = select(User).options(selectinload(User.articles))
users = db.execute(stmt).scalars().all()

背后的 SQL:

-- 第一条:查用户
SELECT * FROM users;

-- 第二条:用 IN 一次查出这些用户的所有文章
SELECT * FROM articles WHERE author_id IN (1, 2, 3, ...);
  • selectinload — 分两条 SQL,用 IN 批量查,不需要 .unique()
  • 适合关联数据量大的情况

5.4 两种预加载怎么选

joinedloadselectinload
SQL 条数1 条(JOIN)2 条(SELECT + IN)
需要 .unique()✅ 需要❌ 不需要
适合场景关联数据少关联数据多
推荐大多数情况关联表数据量大时

六、cascade — 级联操作

控制删除用户时,关联的文章怎么处理。

6.1 默认行为(不设 cascade)

class User(Base):
    articles = relationship("Article", back_populates="author")
    # 没设 cascade
user = db.execute(select(User).where(User.id == 1)).scalars().first()
db.delete(user)
db.commit()
# 结果:
# - 如果 author_id 是 nullable=True → 文章的 author_id 被设为 NULL
# - 如果 author_id 是 nullable=False → 数据库报错!

6.2 设置 cascade="all, delete-orphan"

class User(Base):
    articles = relationship(
        "Article",
        back_populates="author",
        cascade="all, delete-orphan",  # 级联删除
    )
user = db.execute(select(User).where(User.id == 1)).scalars().first()
db.delete(user)
db.commit()
# 结果:用户和他所有的文章都被删除了

6.3 cascade 选项说明

选项含义
all等于 save-update, merge, refresh-expire, expunge(不包含 delete)
delete删除父对象时,删除关联的子对象
delete-orphan子对象从父对象的集合中移除时,也删除它
all, delete-orphan最常用:父删子删,子脱离也删
# "子脱离"是什么意思:
user = db.execute(select(User).where(User.id == 1)).scalars().first()

# 把一篇文章从用户的 articles 列表中移除
article = user.articles[0]
user.articles.remove(article)
db.commit()
# 设了 delete-orphan:这篇文章会被从数据库中删除
# 没设 delete-orphan:这篇文章还在,但 author_id 变成 NULL

七、总结

ForeignKey("users.id")     →  数据库层面:外键约束,保证数据一致性
relationship(...)          →  Python 层面:快捷访问,不用手写查询
back_populates="xxx"       →  双向关联:两端属性名互指
joinedload / selectinload  →  预加载:避免 N+1 查询问题
cascade="all, delete-orphan" → 级联删除:父删子删