第 15 章 基于 ORM 的批量增删改操作
从一个生活场景说起
要给 1000 个员工发年货，有两种方式：
- 逐条办理：每个人分别来人事处，登记、签字、领取，跑 1000 趟；
- 批量办理：把 1000 人的名单一次性交给后勤，按清单统一发放。
逐条 session.add(obj) + 逐条提交，就像第一种方式——每趟都有完整的"接待流程"（ORM 事件、脏检测、往返数据库）。数据量一大，开销触目惊心。本章学习三种"批量办理"的姿势，以及它们各自放弃了什么、换来了什么。
核心概念
SQLAlchemy 的批量操作按"ORM 参与度"由高到低分三档：
| 档位 | 方法 | ORM 参与度 | 速度 |
|---|---|---|---|
| 标准档 | add_all() | 完整（事件、脏检测、回填 id） | 慢 |
| 加速档 | bulk_save_objects() | 部分（跳过部分机制） | 快 |
| 极速档 | session.execute(update/delete/insert) | 零（纯 SQL） | 最快 |
核心权衡：功能 vs 速度。 ORM 机制越少，跑得越快，但对象状态同步、事件钩子等便利也一并失去。
示例代码（逐行讲解）
python
from sqlalchemy import create_engine, update, delete
from sqlalchemy.orm import Session
engine = create_engine(
"mysql+pymysql://root:123456@localhost:3306/demo_db?charset=utf8mb4",
echo=True,
)
with Session(engine) as session:
# ① 批量插入：add_all（走完整 ORM 流程）
session.add_all([User(name=f"用户{i}") for i in range(1000)])
# ② 批量插入：bulk（跳过部分 ORM 机制，更快）
session.bulk_save_objects([User(name=f"用户{i}") for i in range(1000)])
session.commit()
# ③ 批量更新：直接执行 UPDATE，不把对象加载进内存
session.execute(
update(User).where(User.age < 18).values(age=18)
)
# ④ 批量删除：直接执行 DELETE
session.execute(
delete(User).where(User.name.like("临时%"))
)
session.commit()③④ 特别注意写法：update(User) 里传的是映射类，但生成的就是一条赤裸裸的 SQL：
sql
UPDATE user SET age=18 WHERE age < 18
DELETE FROM user WHERE name LIKE '临时%'数据库一条语句搞定全部，根本不产生任何 Python 对象——这是它快的原因，也是它危险的原因（见下文）。
关键知识点
| 方法 | 特点 |
|---|---|
add_all() | 标准 ORM 流程，功能完整但慢 |
bulk_save_objects() | 跳过部分 ORM 逻辑，快 |
session.execute(update/delete) | SQL 层面直接批量操作，最快 |
深入理解：极速档的"失忆症"
SQL 级批量操作有个隐蔽陷阱：它不会同步 Session 中已加载的对象。
python
user = session.get(User, 1) # 加载了张三，age=15
session.execute(update(User).values(age=18)) # 数据库里所有人的 age 都变成了 18
session.commit()
print(user.age) # 可能还是 15！对象没有被刷新数据库已经变了，但你手里的对象还停留在旧值——这就是"脏数据"。对策：
- 批量操作尽早执行（在加载对象之前）；
- 或操作后调用
session.expire_all()强制所有对象过期，下次访问重新查库。
同理，批量 SQL 也不会触发 before_update 之类的 ORM 事件钩子。依赖钩子的业务（如自动记录审计日志）不能用极速档。
进阶内容
1. 分批提交：别让事务变成"巨兽"。 百万级数据一次性 commit，MySQL 的 undo log（回滚日志）会膨胀、锁会长时间不释放，线上直接抖三抖。正确姿势是分批：
python
BATCH = 1000
users = [User(name=f"用户{i}") for i in range(100000)]
for i in range(0, len(users), BATCH):
session.add_all(users[i:i + BATCH])
session.commit() # 每 1000 条提交一次2. MySQL 专属的"存在则更新"。 数据同步场景常见需求："主键已存在就更新，不存在就插入"。MySQL 原生支持 INSERT ... ON DUPLICATE KEY UPDATE，SQLAlchemy 通过方言接口使用：
python
from sqlalchemy.dialects.mysql import insert
stmt = insert(User).values(id=1, name="张三")
stmt = stmt.on_duplicate_key_update(name=stmt.inserted.name)
session.execute(stmt)
session.commit()一条语句搞定"插入或更新"，避免"先查再决定"的两次往返和并发竞争。
3. 怎么选？决策口诀：
- 需要事件钩子、要拿回每条的自增 id →
add_all()； - 纯灌数据、不需要 id →
bulk_save_objects()； - 大批量改/删、没有对象层面的逻辑 → SQL 级
execute(update/delete)。
注意事项
- 批量
update/delete绕过 ORM，不会触发对象的事件钩子，也不会同步 Session 中已加载的对象——操作后 Session 里的旧对象可能是脏数据。 - 批量插入后如果需要拿回每条的主键 id，用
add_all+commit更可靠。 - 超大批量（百万级）建议分批提交（每 1000~10000 条 commit 一次），避免单个事务过大。MySQL 中大事务会撑大 undo log、长时间持锁，影响线上性能。
- MySQL 原生有
INSERT ... ON DUPLICATE KEY UPDATE（冲突时更新），可通过from sqlalchemy.dialects.mysql import insert使用方言特定的insert(...).on_duplicate_key_update(...)，适合"存在则更新，不存在则插入"的批量同步场景。 - 批量操作同样遵守第 5 章的铁律：先 select 验证条件，再执行，并检查影响行数。
本章小结
- 三档批量操作：
add_all（功能全）→bulk_save_objects（快）→ SQL 级execute（最快）； - 速度与 ORM 特性成反比：越快，事件钩子和对象同步越少；
- SQL 级批量后记得
expire_all()防脏数据； - 百万级数据分批 commit，保护 MySQL。