第 14 章 基于 ORM 的查询操作

从一个生活场景说起

前面几章的查询都比较"朴素"：按名字找人、按主键取卡。真实业务的查询更像在人才市场筛选简历：

• "年龄在 25 到 35 之间"（范围过滤）；
• "姓张的，或者姓李的"（多条件组合）；
• "学历这一栏不能是空的"（判空）；
• "按工作年限从高到低排，第 3 页，每页 10 人"（排序分页）；
• "每个年龄段有多少人？"（分组统计）；
• "顺便把每个人的证书清单也一起拿来"（关联预加载）。

本章就是把这套"筛简历话术"全部翻译成 SQLAlchemy。

核心概念

本章系统梳理 ORM 查询的常用套路。基础入口都是 select(Model) + session.scalars()——把它当成一句咒语记牢，剩下的只是往里填条件。

常用查询示例（逐组讲解）

from sqlalchemy import create_engine, select, and_, or_, func
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:
    # 1. 等值 / 比较 —— 最基础的筛选
    select(User).where(User.age == 20)
    select(User).where(User.age > 18)

    # 2. 多条件 AND / OR —— 组合筛选
    select(User).where(and_(User.age > 18, User.name.like("张%")))
    select(User).where(or_(User.name == "张三", User.name == "李四"))
    # 也可以直接连续 where（AND 语义），效果与 and_ 相同
    select(User).where(User.age > 18).where(User.name.like("张%"))

    # 3. IN / 模糊 / 判空 —— 三种高频筛选器
    select(User).where(User.name.in_(["张三", "李四"]))
    select(User).where(User.name.like("%三%"))
    select(User).where(User.age.is_(None))       # IS NULL：年龄栏为空
    select(User).where(User.age.is_not(None))    # IS NOT NULL

    # 4. 排序与分页 —— 多列排序 + 翻页
    select(User).order_by(User.age.desc(), User.id.asc()).limit(10).offset(20)

    # 5. 聚合统计 —— "有多少""平均多少"
    total = session.scalar(select(func.count()).select_from(User))
    avg_age = session.scalar(select(func.avg(User.age)))

    # 6. 分组 —— "每个年龄段各多少人"
    stmt = select(User.age, func.count()).group_by(User.age).having(func.count() > 1)

    # 7. 关联查询（JOIN）—— 跨表筛选：找出写过"教程"类文章的用户
    stmt = select(User).join(User.articles).where(Article.title.like("%教程%"))

    # 8. 预加载关联，避免 N+1 问题 —— 查用户时把文章一并取来
    from sqlalchemy.orm import selectinload
    stmt = select(User).options(selectinload(User.articles))
    users = session.scalars(stmt).all()
    for u in users:
        print(u.name, len(u.articles))   # 不再触发额外 SQL

逐组说明：

• 第 1~3 组：都是 .where() 的"筛子"。like("张%") 中的 % 是通配符，表示"任意长度的任意字符"，所以 "张%" 匹配所有姓张的；
• 第 4 组：order_by 可以传多列，先按年龄降序、年龄相同的再按 id 升序；limit/offset 就是"每页几条、跳过前几条"；
• 第 5 组：func 是 SQL 函数的"万能入口"，func.count()、func.avg() 等通过它调用；
• 第 6 组：group_by 把同年龄的行"捆成一捆"，func.count() 数每捆有几行，having 再筛掉人数 ≤1 的捆；
• 第 7 组：.join(User.articles) 利用第 11 章定义的 relationship 自动生成 JOIN 条件，不用手写 user.id == article.author_id；
• 第 8 组：本章最重要的性能技巧，下面详谈。

关键知识点

功能	写法
多条件	and_() / or_() / 连续 where()
IN	.in_([...])
模糊	.like() / .ilike()（不区分大小写）
判空	.is_(None)
聚合	func.count() func.avg() func.sum()
分组	.group_by() + .having()
预加载	selectinload() / joinedload()

深入理解：N+1 问题——ORM 的头号性能杀手

看这段"人畜无害"的代码：

users = session.scalars(select(User)).all()   # 1 条 SQL：查 100 个用户
for u in users:
    print(u.articles)                          # 每个用户触发 1 条 SQL 查文章

100 个用户 = 1 + 100 = 101 条 SQL。这就是著名的 N+1 问题：1 条主查询 + N 条懒加载查询。数据库连接再多也经不起这么折腾。

解药是预加载（Eager Loading）——查询主表时就把关联数据一并取来：

from sqlalchemy.orm import selectinload

stmt = select(User).options(selectinload(User.articles))
users = session.scalars(stmt).all()

selectinload 的原理很聪明：查完用户后，收集所有用户 id，再发一条 SELECT * FROM article WHERE author_id IN (1,2,3,...,100)，然后在 Python 里自动配对。101 条 SQL 变成 2 条。

另一种预加载是 joinedload：直接 JOIN 出来只用 1 条 SQL。选择经验："多"的一方用 selectinload（避免 JOIN 产生的行数膨胀），"一"的一方用 joinedload（一对一/多对一 JOIN 不膨胀）。

深入理解：where 与 having 的分工

这两个都是"筛子"，但作用的阶段不同：

SELECT age, COUNT(*) FROM user
WHERE age >= 18        -- 第一阶段：先筛原始行
GROUP BY age           -- 第二阶段：把剩下的行捆成组
HAVING COUNT(*) > 1    -- 第三阶段：再筛组

口诀：where 筛行，having 筛组。把过滤条件放错位置，要么结果不对，要么白白多扫描数据。

进阶内容

1. 只查部分列。 不需要整个对象时，直接 select 列，性能更好：

stmt = select(User.name, User.age).where(User.age > 18)
rows = session.execute(stmt).all()   # 注意：返回的是 Row 而不是 User 对象

2. 去重与计数组合。

# 有多少个不同的年龄值
session.scalar(select(func.count(func.distinct(User.age))))

3. MySQL 的大小写问题。 MySQL 默认排序规则（utf8mb4_unicode_ci）下，like 是不区分大小写的：name = 'Tom' 能匹配 'tom'。需要严格区分时，可以在列定义时用区分大小写的排序规则（如 utf8mb4_bin），或使用二进制比较。

4. 前缀通配的索引陷阱。 like("张%") 能用上索引，like("%三%")（前面也有 %）则索引失效、全表扫描。十万行以下的表无所谓，百万级就要考虑全文索引或搜索引擎了。

注意事项

• 判断空值必须用 .is_(None)，不能用 == None（会被翻译成 = NULL，永远查不到）。
• N+1 问题是 ORM 性能头号杀手：查 100 个用户后逐个访问 user.articles，会额外发 100 条 SQL。用 selectinload 一次预加载。
• .having() 是对分组结果的过滤，.where() 是对原始行的过滤，别混淆。
• MySQL 中 like 默认不区分大小写（utf8mb4_unicode_ci 排序规则下），需要区分时改用区分大小写的排序规则或二进制比较。
• 模糊查询 like("%xxx%") 前缀通配会导致索引失效，大数据量下慎用。
• 聚合查询（count/avg）返回的是单个值，用 session.scalar() 取，不要用 scalars().all()。

本章小结

• 万能入口：select(Model) + session.scalars()；
• 筛选三件套：where（行）、group_by + having（组）、order_by + limit（排序分页）；
• N+1 是 ORM 性能第一课，列表场景记得 selectinload；
• like 的通配符位置和大小写规则，在 MySQL 里都有讲究。