数据库查询优化

核心思想

减少与数据库的通信次数（减少往返），以及减少每次通信的数据量（瘦身）

1. 著名的 “N+1 问题” 与预加载（Relationship Loading）

这是所有数据库框架中最经典的性能黑洞。

  • 大白话场景： 你要打印 10 个班级的学生名单。
    • 脑残做法（N+1）： 你先去办公室查到了这 10 个班级（1次查询）。然后，你回班级拿第1班的名单，再去办公室；拿第2班的名单，再去办公室……为了 10 个班级，你一共跑了办公室 $10 + 1 = 11$ 次。如果有一万个班级，你的腿就废了（数据库卡死）。
    • 聪明做法（预加载）： 你在查 10 个班级的时候，直接跟办公室老师说：“把这 10 个班级里的学生名单顺便一次性都打包给我！”（只跑 1次 办公室）。

SQLAlchemy 的官方解法：
在查询时，使用 options() 明确指定加载策略。
官方提供了三种武器：
武器 A：joinedload()（连表预加载）
原理： 用 SQL 的 LEFT OUTER JOIN 把两张表拼成一张大表，一次性全拿回来。
适用场景：
一对一（如用户和他的身份证信息）或者数据量不大的关联。

1
2
3
from sqlalchemy.orm import joinedload
    # 一次性把用户和他们的订单全部用一条 SQL 查出来
    stmt = select(User).options(joinedload(User.orders))


武器 B：selectinload()（分批预加载 - 官方最推荐）
原理： 先用 1 条 SQL 查出所有班级，再发 1 条 SQL 查出“所有属于这些班级的学生”（用 IN (班级ID列表)）。一共雷打不动就 2 条 SQL。
适用场景： 一对多（比如班级和学生、文章和评论），这是 2.0 文档中最推荐的默认行为。
1
2
from sqlalchemy.orm import selectinload
stmt = select(User).options(selectinload(User.orders))

2. 延迟加载大字段（Deferred Columns）

大白话场景： 你的商品表里有一列是“商品详情描述”（包含了大量字数和图片 HTML）。当用户在后台看“商品列表”（只需要看商品名和价格）时，你把几万字的详情也全部从数据库里捞出来了。这就像运送一箱苹果，你非要把装苹果的集装箱卡车也一起抬走，极其浪费内存和网络带宽。
优化策略： 告诉数据库，平时别查这一列，只有当我真正点进商品详情、调用 product.description 时，你再去数据库里帮我拿。
SQLAlchemy 写法：在模型定义或查询时将其设置为 defer（延迟加载）：

1
2
3
from sqlalchemy.orm import defer
# 查询用户时，故意把“用户大头像（avatar）”这种大字段漏掉，提升查询速度
stmt = select(User).options(defer(User.avatar))

3. 只捞需要的列，坚决不用 SELECT *（Yielding Columns）

大白话场景： 去饭店吃饭，你只想知道今天有没有“红烧肉”（True/False），结果老板把一整盘红烧肉做出来端到你面前让你看。
优化策略：
不要总是习惯性地把整个对象（User）捞出来。如果你只需要用户的 ID 和名字，就只查这两列。这样 SQLAlchemy 就不用费尽心思把数据组装成一个复杂的 Python 对象，速度飞快。
SQLAlchemy 写法：

1
2
3
4
# 差劲做法：查出整个用户对象
user = session.scalars(select(User).where(User.id == 1)).first()
# 优秀做法：只拿名字和邮箱，返回的是一个干净的元组（Tuple）
result = session.execute(select(User.name, User.email).where(User.id == 1)).first()

4. 批量插入与更新（Bulk Operations）

大白话场景： 你要把 10000 个快递送进速递易快递柜。
慢速做法：
刷一次卡，放一个快递，关门；再刷一次卡，放一个，关门……重复一万次，后面排队的人想打人。

高速做法：
管理员直接打开柜子的后备大门，哗啦一下把一万个快递全部塞进去，一次性锁门。

SQLAlchemy 写法：
在 2.0 文档中，官方极大优化了批量操作。当你需要插入成千上万条数据时，绝对不要用 for 循环去 session.add()，而是使用 insert() 的批量语法：

1
2
3
# 批量插入一万条数据，速度提升几十倍
data_list = [{"name": f"user_{i}", "email": f"{i}@qq.com"} for i in range(10000)]
session.execute(insert(User), data_list)

5. 外部执行与流式查询（Server-side Cursors / Yield Per）

大白话场景： 数据库里有 1000 万条用户数据，你要把它们全部导出来做分析。如果你直接执行 select(User)，数据库会把 1000 万条数据一次性全部塞进你的 Python 内存里，你的服务器当场就会因为内存溢出（OOM）而崩溃死机。
优化策略：
像吸管吸珍珠奶茶一样，一次只吸 1000 条，处理完了，再吸下 1000 条。
SQLAlchemy 写法：
使用 yield_per() 分批次从数据库服务器拉取数据，内存永远保持在极低水平：

1
2
3
4
# 每次只从数据库拉取 1000 条数据到内存中，处理完再拿下一批
stmt = select(User).execution_options(yield_per=1000)
for user in session.scalars(stmt):
    print(user.name)

6. 查询缓存（Compiled Cache）

大白话场景： 你每次问学霸：“$12345 \times 67890$ 等于多少？”，学霸都要拿草稿纸算半天（数据库解析 SQL 语句）。如果学霸聪明，他会把算过的题目和答案记在脑子里，下次你再问一模一样的题目，他一秒钟就能报出答案。
官方机制：
SQLAlchemy 2.0 默认自带了 Lambda 语句缓存和 Compiled Cache。它会把你写的 Python 查询代码翻译成 SQL 的过程缓存起来。
小白注意点： 为了让这个缓存生效，尽量不要在查询语句中动态拼接字符串，而是使用参数化查询（SQLAlchemy 默认的 .where(User.id == my_id) 就是参数化的，千万不要自己去拼写类似 f”SELECT * FROM user WHERE id = {my_id}” 的高危垃圾代码）。

总结：你的性能优化检查清单（Checklist）

以后写完代码，对着这个清单瞅一眼：

  • 涉及两张表一起查了吗？（用了 selectinload 或 joinedload 吗？如果没有，小心 N+1 毁灭你的系统）。
  • 这一列是不是有上万字的超级大文本？（用了 defer 延迟加载吗？）。
  • 我是不是只需要一个数字或一个名字？（有没有偷懒写成了 select(User)？改成 select(User.name) 吧）。
  • 我是不是在用循环往数据库里疯狂插数据？（快改成 session.execute(insert(User), list) 批量模式）。