流式读取用yield_per()配合stream_results=True避免OOM；批量插入用bulk_insert_mappings()跳过ORM开销；防N+1用selectinload()预加载；复杂查询用with_entities()或text()绕过ORM。
用 yield_per() 流式读取大结果集
查几万行数据时，session.query().all() 会一次性把所有对象加载进内存，容易 OOM。这时候得换流式读取：yield_per() 让 SQLAlchemy 按批拉取、逐批返回，避免全量缓存。
注意它只对查询生效，且必须配合 execution_options(stream_results=True) 才真正启用底层流式（尤其 PostgreSQL/MySQL）：
query = session.query(User).filter(User.active == True)
for user in query.execution_options(stream_results=True).yield_per(1000):
process(user) # 每次只 hold 1000 个对象常见坑：
-
yield_per()后不能调用.count()或.first()—— 会触发完整执行并清空游标 - SQLite 不支持流式，
stream_results=True会被静默忽略 - 如果后续要修改这些对象并
commit()，记得在每批处理完后调用session.expunge_all()，否则旧对象持续占用内存和 identity map
批量插入用 bulk_insert_mappings() 而非循环 add()
往表里插几千条记录，用 session.add() 循环加再 commit()，性能极差：每条都走 ORM 映射、事件触发、脏检查。换成原生级批量操作，快一个数量级以上。
立即学习“Python免费学习笔记（深入）”；
bulk_insert_mappings() 直接构造 SQL INSERT 多值语句（如 INSERT INTO t (a,b) VALUES (?,?),(?,?)），跳过 ORM 开销：
data = [{"name": "A", "email": "a@x.com"}, {"name": "B", "email": "b@x.com"}]
session.bulk_insert_mappings(User, data)
session.commit()关键限制：
- 不触发
@validates、before_insert等 ORM 事件 - 不会填充主键（如自增 ID）回对象 —— 返回值是空列表，别指望它赋值
user.id - 不能混用未映射字段；字段名必须严格匹配模型定义（大小写敏感）
- PostgreSQL 下若含 JSON 字段，需确保传入的是 Python dict/list，不是字符串
避免 N+1 查询：用 joinedload() 或 selectinload() 预加载关联
查用户列表再逐个访问 user.posts，就是典型 N+1：1 次查用户 + N 次查 posts。ORM 默认懒加载，不显式声明就掉坑里。
两种主流预加载方式：
-
joinedload()：用 JOIN 一条 SQL 拉取主表 + 关联表，适合关联数据量小、且不重复（如用户+头像） -
selectinload()：先查主表 ID 列表，再用WHERE id IN (...)查关联表 —— 更稳定，尤其一对多时避免笛卡尔爆炸，推荐优先用
示例：
# 错误：N+1
users = session.query(User).limit(100).all()
for u in users:
print(u.posts) # 每次都发新查询
<h1>正确：一次查完</h1><p>users = session.query(User).options(selectinload(User.posts)).limit(100).all()注意：lazy='select'（默认）只是开关，不等于自动优化；必须显式加 options() 才生效。
复杂过滤改用 Query.with_entities() 或原生 text()
当查询只取几个字段、带复杂计算或数据库特有函数（比如 PostgreSQL 的 ts_rank、MySQL 的 JSON_EXTRACT），硬套 ORM 模型反而拖慢且难写。
两种轻量替代：
-
with_entities()：跳过模型实例化，直接返回元组或命名元组，省掉 ORM 映射开销 -
text()：写原生 SQL 片段，配合bindparam防注入，适合高度定制场景
例如统计活跃用户的邮箱域名分布：
from sqlalchemy import text
<p>stmt = text("SELECT SUBSTRING_INDEX(email, '@', -1) as domain, COUNT(*) FROM users WHERE active = 1 GROUP BY domain")
results = session.execute(stmt).fetchall() # 返回普通 tuple 列表小心点：
-
with_entities()返回的是KeyedTuple，不是模型对象，不能调.save()或访问关系属性 -
text()中的参数必须用:name占位符，不能用 Python 字符串格式化，否则 SQL 注入 - 跨数据库移植性归零 —— 写了
ts_rank就别想轻松切到 SQLite
批量加速不是堆技巧，而是根据数据量、字段需求、关联深度，选对那条“最短路径”。流式、批量、预加载、绕过 ORM —— 每种都有明确适用边界，用错地方反而更慢。
