#一对多关系
最常用的关系类型:一个用户有多篇文章,一篇文章属于一个用户。
#一、场景说明
用户表 (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 两种预加载怎么选
joinedload | selectinload | |
|---|---|---|
| SQL 条数 | 1 条(JOIN) | 2 条(SELECT + IN) |
需要 .unique() | ✅ 需要 | ❌ 不需要 |
| 适合场景 | 关联数据少 | 关联数据多 |
| 推荐 | 大多数情况 | 关联表数据量大时 |
#六、cascade — 级联操作
控制删除用户时,关联的文章怎么处理。
#6.1 默认行为(不设 cascade)
class User(Base):
articles = relationship("Article", back_populates="author")
# 没设 cascadeuser = 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" → 级联删除:父删子删
