SQL 表达式与数据操作

本章你将学会

• select() 骨架与链式拼装：where / order_by / limit / offset
• 比较运算符全家桶：== 的真面目、like vs ilike、in_、is_(None)
• and_ / or_ / not_ 组条件，& | ~ 的括号陷阱
• func 聚合：count / sum / avg + group_by / having
• update().values()（含 synchronize_session）与 delete().where()
• join / outerjoin、外键一对多、多对多 any() / contains() 查询
• execute / scalars / scalar_one 的返回值差异
• 为什么 Core 表达式天然防 SQL 注入

第 04 章用 text() 手写 SQL、第 05 章把五张表立了起来，当时都留了同一句话：SQL 表达式第 06 章细讲。现在兑现。这一章是纯用法手册：每个 API 按「作用 → 签名 → 最小示例 → 常见坑」过一遍，学完你就能把数据库当 Excel 一样盘。

本章配套 demo（aiosqlite 上实测通过）： - demo/demo_select.py：查询全家桶——select / where / 排序分页 / 逻辑组合 / join / 关系条件 / 预加载 - demo/demo_write.py：写操作——update / delete / 聚合分组 / 外键级联

示例业务统一用「墨集 inkhub 内容平台」：Channel 栏目、Article 文章（channel_id 外键）、Tag 标签、article_tag 关联表、ArticleVersion 版本（version_no / word_count / status）——第 05 章建好的那五张表，本章只管用。

1. 心智模型：先拼积木，再交翻译

先看同一件事的两种写法：

from sqlalchemy import select, text

# 写法一：text() —— SQL 已经是成品字符串，写歪了只有数据库知道
stmt = text("SELECT * FROM article WHERE channel_id = 1")

# 写法二：Core 表达式 —— 用 Python 对象拼 SQL，每一步都被检查
stmt = select(Article).where(Article.channel_id == 1)

打个比方：text() 是捏橡皮泥——一整块捏成形，裂了歪了没人提醒；Core 是搭乐高——列、运算符、函数全是标准件，凸点对凹点，装错位置根本插不进去。把拼好的作品打印出来看看：

print(select(Article.id, Article.title).where(Article.channel_id == 1))
# SELECT article.id, article.title
# FROM article
# WHERE article.channel_id = :channel_id_1

三个细节值得盯一眼：

1. Article.channel_id == 1 不是在「比较」，是在「构造」——列的运算符被重载，返回一个 SQL 表达式对象；
2. 值 1 被自动装进绑定参数 :channel_id_1，和 SQL 骨架分道扬镳——这是第 13 节「天然防注入」的根基；
3. 语句只是 Python 对象：不执行、不发网络，可以打印检查、可以反复执行。

一句话记住本章的世界观：构建与执行分离。你负责拼乐高，引擎负责按数据库的方言翻译——同一份 limit(10)，PG、SQLite、MySQL 各拿各的说明书去执行。

2. select()：查询的骨架

作用：创建 SELECT 语句，决定「查什么」。

签名：select(*entities)——entities 可以是模型类、列、聚合表达式，一个或多个。

最小示例——三种典型查法：

select(Article)                                # 整个实体：配合 scalars() 拿模型对象
select(Article.id, Article.title)              # 挑列查：拿到一行行的 Row
select(Channel.name, func.count(Article.id))   # 混入聚合列：第 7 节

链式拼装（每个方法返回语句的变形，随便往后接）：

stmt = (
    select(Article)
    .where(Article.channel_id == 1)          # 第 4 节
    .order_by(Article.created_at.desc())     # 第 6 节
    .limit(10)
)

常见坑

• select(Article) 和 select(Article.title) 查出来的东西类型不同：前者是模型对象流，后者是一行行的值——用混了取值方式，拿到的数据「能看不能摸」。第 3 节专门掰扯。
• 多列 select 别接 .scalars()：它只抽第一列，其余列全丢。
• 拼进 select 的模型必须已定义且被 import——第 05 章「档案柜」的规矩：没登记的图纸，这里直接 NameError。

3. 取结果：execute / scalars / scalar_one

第 04 章见过 execute + scalars() 的组合拳，这里把 Result 的家底一次摊开。先记这张层次图：

await session.execute(stmt)   →  Result          一行一个 Row（带列名的元组）
        .scalars()             →  ScalarResult   只保留每行第一列
        .scalar_one()          →  第一行第一列，且行数必须恰好 1

最小示例——同一个查询的三种取法：

async with session_factory() as session:
    # ① execute：拿 Result，迭代出一行行的 Row（支持元组解包）
    result = await session.execute(
        select(Article.id, Article.title).where(Article.channel_id == 1)
    )
    for id_, title in result:
        print(id_, title)

    # ② session.scalars：直接要「第一列」流——实体查询即模型对象
    articles = (await session.scalars(select(Article))).all()

    # ③ scalar_one：单值，必须恰好一行一列
    title = (
        await session.execute(select(Article.title).where(Article.id == 2))
    ).scalar_one()

拆快递的比喻：execute 收到一整箱，每层一个 Row 包装；scalars() 是开箱直取——每层直接抽出第一件；scalar_one() 最苛刻——必须恰好一箱一层一件，多一件少一件当场报警。

取值方法矩阵（Result 与 ScalarResult 同名方法，行为一致）：

常见坑

• 0 行、多行对 one() 家族是异常而不是返回 None：scalar_one() 查出两行直接抛 MultipleResultsFound——这不是它在找茬，是在帮你兜住「WHERE 条件不够唯一」的 bug。允许查不到的场景才用 one_or_none() / first()。
• Result 是一次性流：for 循环跑完再 .all()，得到的是空——结果要复用，第一次就存进变量。
• conn.execute(stmt) 和 session.execute(stmt) 返回的都是 Result；区别是会话版附带 ORM 的增值服务（scalars()、关系加载、autoflush）。

4. where()：条件与比较运算符

作用：给语句挂 WHERE，逐行筛选。

签名：where(*条件)——塞多个参数就是 AND（第 5 节展开），每个参数是一个布尔表达式对象。

运算符对照表（左 SQL，右 Core 写法）：

头号坑：== 的真面目

Article.channel_id == 1 看着像比较，实际是构造——列对象重载了 __eq__，返回一个 SQL 表达式。两个直接后果：

# ① 单等号是赋值，Python 编译期直接拦（SyntaxError），这坑反而不常踩
select(Article).where(Article.channel_id = 1)

# ② 真正的坑：拿它当布尔判断——表达式对象永远 truthy
if Article.status == "draft":    # 恒为 True！
    ...

写惯 Python 的人在第 ② 种翻车：判断「这篇文章的状态」应该用查出来的实例 article.status，而不是类属性表达式 Article.status == "draft"。前者是数据，后者是 SQL 零件——零件永远不为假。

like vs ilike：性格随数据库的一对兄弟

• like：是否区分大小写看数据库心情。PG 区分；SQLite 的 LIKE 对 ASCII 不区分（中文不受影响）。同一行代码换个库行为就变——「本地好好的、上线查不对」的经典来源。
• ilike：明确不区分。PG 上生成原生 ILIKE；SQLite 这类没有 ILIKE 的库，自动降级编译成 lower(title) LIKE lower(?)。

结论：语义上要「确定不区分大小写」就写 ilike，别赌 like 的方言行为。

顺带认识 like 的快捷亲戚：contains("入门")、startswith("Fast")、endswith("指南")——自动帮你拼两边的 %，读起来也更像人话。

in_() 的坑：列表常来自用户勾选，可能是空的。in_([]) 会被编译成 1 != 1 式的恒假子句（SQLAlchemy 2.0 的安全行为），一行都匹配不到；但业务语义要想清楚——「一个都没勾」是查不到东西，还是干脆跳过这个条件？前者保留 in_，后者判断空列表后别拼这个 where。真正危险的是拿 text() 手拼 IN ({...})，那是往注入的老路上走。

is_(None) 的坑：Article.summary == None 其实能跑（会被编译成 IS NULL），但 == None 会被 flake8 按 E711 拦下——Python 惯例与 None 比较用 is，写成 == 也容易让读代码的人误会。统一 is_(None) / is_not(None)，谁看了都不多想。

5. and_ / or_ / not_：条件的布尔运算

作用：把多个条件按与、或、非组合。

签名：and_(*expr)、or_(*expr)、not_(expr)，可任意嵌套。

最小示例——「标题带 fastapi」或「深度长文且写了摘要」的文章：

from sqlalchemy import and_, not_, or_

stmt = select(Article).where(
    or_(
        Article.title.ilike("%fastapi%"),
        and_(
            Article.channel_id == 2,           # 深度长文
            not_(Article.summary.is_(None)),   # 且写了摘要
        ),
    )
)

where 的隐藏语法糖：多个参数就是 AND——

.where(Article.channel_id == 2, Article.summary.is_not(None))
# 等价于
.where(and_(Article.channel_id == 2, Article.summary.is_not(None)))

& | ~ 的括号陷阱

AND / OR / NOT 也能用位运算符写。但 Python 的位运算优先级高于比较运算符，所以：

# ❌ 想表达 (channel_id == 1) | (id == 2)
# 实际解析成 Article.channel_id == (1 | Article.id) == 2 ——当场 TypeError
.where(Article.channel_id == 1 | Article.id == 2)

# ✅ 用位运算符必须层层加括号
.where((Article.channel_id == 1) | (Article.id == 2))

记不住优先级就老实用 and_ / or_ / not_ 函数风格——永不歧义，本章 demo 全是这种写法。

6. order_by / limit / offset：排序与分页

作用：排序 + 截取，列表页接口的标配三件套。

签名：

stmt = (
    select(Article)
    .order_by(Article.created_at.desc(), Article.id)   # ① 多键排序
    .limit(10)        # ② 最多取 10 行
    .offset(20)       # ③ 跳过前 20 行
)

• asc() 是默认方向，desc() 倒序；多个键按书写顺序依次生效；
• 分页公式：offset = (page - 1) * per_page——第 3 页、每页 10 条，就是 limit(10).offset(20)；
• 这俩是方言适配的代表案例：PG / SQLite / MySQL 用 LIMIT ... OFFSET ...，SQL Server 得换 OFFSET FETCH——写 Core 的你不用管，引擎按方言翻译。

常见坑

• 不带 order_by 就分页 = 抽盲盒。SQL 标准压根没承诺无 ORDER BY 时的返回顺序，数据库今天一个心情明天一个心情——第 1 页和第 2 页可能重叠、可能漏行。
• 排序键不唯一 = 翻页会抖。两篇文章 created_at 相同，谁排前面由数据库随手定——翻页时数据忽上忽下。药方：末尾补一个唯一列兜底，.order_by(Article.created_at.desc(), Article.id)。像运动会排名：成绩相同看编号，永远分得出高下。
• 深分页的性能账。offset(100000) 意味着数据库老老实实扫过并扔掉前 10 万行。后台系统翻到几万页越来越慢，元凶就是它——解法是游标分页（where(Article.id > 上一页最后一条).limit(20)），进阶话题，先记住症状和方向。

7. func：聚合与分组报表

作用：func 是 SQL 函数的万能转接台——func.函数名(参数) 能拼出任意函数调用，聚合（count / sum / avg / min / max）是它的头号业务。

最小示例——各栏目的文章数：

from sqlalchemy import func

stmt = (
    select(Channel.name, func.count(Article.id).label("articles"))
    .outerjoin(Article, Channel.id == Article.channel_id)   # 空栏目也统计（第 8 节）
    .group_by(Channel.name)                                 # 按栏目分桶
)
async with session_factory() as session:
    for name, cnt in await session.execute(stmt):   # Row 也能直接解包
        print(name, cnt)

label() 给聚合列起名：Row 按名字取值（row.articles），比 row[1] 可读一个量级。

再上一档：sum / avg 多级统计——用 05 章 ArticleVersion 的 word_count / status 正好：

stmt = (
    select(
        ArticleVersion.status,
        func.count().label("versions"),                      # count(*) 数行数
        func.sum(ArticleVersion.word_count).label("total"),  # 总字数
        func.avg(ArticleVersion.word_count).label("avg"),    # 平均字数
    )
    .group_by(ArticleVersion.status)
)

having：给「组」加条件

stmt = stmt.having(func.count() >= 2)   # 只保留版本数 ≥ 2 的状态

where 和 having 的分工，用招聘打比方：where 是初筛简历——不合格的人根本进不了统计；having 是按部门评优——先组队算完绩效，再淘汰整个部门。where 过滤行、发生在分组前；having 过滤组、发生在分组后——所以 having 里能用聚合值当条件，where 里不行。

常见坑

• select 了非聚合列却不 group_by 它：PG 直接报错（must appear in the GROUP BY clause）；SQLite 宽松放行、随便挑一行凑数。于是「本地好好的、上生产炸了」。规矩：select 里每个非聚合列都写进 group_by。
• count(列) ≠ count(*)：count(Article.summary) 不统计该列为 NULL 的行。数「没写摘要的文章」用 count() - count(Article.summary)；数总行数用 count()（或主键列）。
• count 永远返回一行：空表时它返回一行、值是 0，而不是零行——判断「有没有数据」看值，别看 first() is None。
• avg 的返回类型看方言：SQLite 给 float；PG 对整数列求 avg 返回 Decimal。做除法或格式化前先 float(...) 兜一道，两种环境行为才一致。

8. join / outerjoin：把两张表缝起来

作用：连接查询，让两表的行按条件两两配对。

签名：.join(target, onclause=None)——target 是模型或表；onclause 可省（ORM 能按外键 / relationship 自动推断）。.outerjoin() 是它的左外连接版。

最小示例——文章配栏目名：

# 写法一：显式 on 条件
stmt = (
    select(Article.title, Channel.name)
    .join(Channel, Article.channel_id == Channel.id)
)

# 写法二：交给 relationship 自动推断（05 章建过 Article.channel 就能省）
stmt = select(Article.title, Channel.name).join(Article.channel)

谁是左表？select 的第一个实体说了算：

select(Article.title, Channel.name).join(Channel, ...)   # FROM article JOIN channel
select(Channel.name, Article.title).join(Article, ...)   # FROM channel JOIN article

想另立主语用 select_from()：select(Channel.name).select_from(Article).join(Article.channel)——FROM 从 article 起步。

outerjoin：左边的行一个都不丢

stmt = select(Channel.name, Article.title).outerjoin(
    Article, Channel.id == Article.channel_id
)
# 没有文章的栏目照常出现，文章列补 NULL

内连接像拉链两边齿齿咬合——配不上的行直接消失；左外连接是只保左排的齿——右齿缺位用 NULL 填空。统计「含空栏目」的报表（第 7 节的按栏目计数）必须用它，否则空栏目凭空消失。

常见坑

• 两表的列都 select 了却忘了 join：生成笛卡尔积，行数 = 左行数 × 右行数。症状是结果暴涨、每条数据都似曾相识。看到重复组合，先查 join。
• join(目标) 省略 onclause 的前提是外键能唯一推断：两表间有多个外键或不搭界时会直接抛错——老老实实把 onclause 写上。
• join 方向写反，查询逻辑就镜像了：Article.channel 与 Channel.articles 生成的是两条方向相反的 SQL——写之前想清楚「FROM 谁出发」。

9. 关系当条件：has() / any() / contains()

join 是手动挡；SQLAlchemy 还允许把 relationship 直接写进 where——join 自动补好。05 章建的关系，从这节开始回本。

口诀：单对象端用 has()，集合端用 any() / contains()。（拿单数属性 → has；拿复数属性 → any / contains）

# has()：多对一端——「文章的栏目」满足条件（自动 JOIN channel）
select(Article).where(Article.channel.has(Channel.slug == "news"))

# any()：集合端——「文章的标签」中存在满足条件的（自动 JOIN article_tag + tag）
select(Article).where(Article.tags.any(Tag.name == "Python"))

# contains()：集合端，入参是具体对象——标签集合里含有这个 Tag
select(Article).where(Article.tags.contains(python_tag))

any() 吃条件表达式，contains() 吃已入库的对象——后两句查出来是同一批文章，只是入参姿势不同。反方向照样能用：

# 某篇文章都贴了哪些标签：Tag 的文章集合包含这篇文章
select(Tag).where(Tag.articles.contains(article_obj))

多对一端还有个语法糖——外键列直接比对象：

# 都翻译成 WHERE article.channel_id = :channel_id
select(Article).where(Article.channel == channel_obj)
select(Article).where(Article.channel_id == channel_obj.id)

常见坑

• contains() 的对象必须已持久化（有主键）：传一个刚 Tag(name="x") 的内存对象，主键还是 None，条件无从定位行——先 add + flush 再用。
• has() / any() 用反端会直接抛错。别死记，按语义想：has 问的是「那一个对象满足吗」，any 问的是「这堆里有没有满足的」。
• 关系条件查出的实体，关系属性默认仍是未加载的——article.tags 想直接访问，还得靠下一节的预加载。

10. 预加载补课：selectinload / joinedload

第 05 章挖的坑这里填上：异步会话里访问懒加载关系会抛 MissingGreenlet——SQLAlchemy 想在你读属性那一刻悄悄发 SQL，而 asyncio 容不下这种「背着你做 IO」。

作用：.options() 声明预加载，把关系的加载并进你看得见的查询计划。

from sqlalchemy.orm import selectinload

stmt = select(Article).options(
    selectinload(Article.channel),   # 多对一
    selectinload(Article.tags),      # 多对多
)
articles = (await session.scalars(stmt)).all()
for a in articles:
    print(a.channel.name, [t.name for t in a.tags])   # 已装好，随便读

两位主角的分工（经验法则）：

常见坑

• 报错名先认下：sqlalchemy.exc.MissingGreenlet——看到它，八成是忘了 options。注意 options 只装列出来的关系，访问没列的还是懒加载、还是炸。
• 分页 + 集合关系的组合，社区习惯用 selectinload：joinedload 套集合时 SQLAlchemy 会自动包一层子查询来保证 LIMIT 语义正确（结果不会错），但 SQL 明显变绕；selectinload 的两条直白 SQL 更好读也好调。

11. update().values()：更新

作用：构造 UPDATE 语句，按条件批量改列值。

签名：update(实体).where(条件).values(列=值, ...)。

最小示例——连接级（Core 最纯粹的形态）：

from sqlalchemy import update

async with engine.begin() as conn:
    result = await conn.execute(
        update(ArticleVersion)
        .where(ArticleVersion.status == "review")
        .values(status="published")
    )
    print(result.rowcount)   # 实际改动了几行

rowcount 是 UPDATE / DELETE 的「战报」，写接口时直接当业务数字用。

values 的两种给法：

# ① Python 值：渲染成绑定参数 :status_1
.values(status="published")

# ② SQL 表达式：渲染成 article_version.word_count + 100，数据库端计算
.values(word_count=ArticleVersion.word_count + 100)

第 ② 种常被忽略——想「全部加 100」，不是查出所有行在 Python 里逐个加，而是把表达式交给数据库一条 SQL 搞定（一百万行也只有一条语句）。

synchronize_session：会话里的对象跟不跟着变

在会话里执行 Core update（await session.execute(update(...))）时有个绕不开的问题：数据库的行变了，会话里已经加载过的对象怎么办？这正是 synchronize_session 管的事：

对比实验（demo_write.py 第 2 节实测）：

async with session_factory.begin() as session:
    version = (await session.scalars(select(ArticleVersion).where(...))).one()

    await session.execute(
        update(ArticleVersion).where(...).values(status="archived")
    )
    print(version.status)        # 'archived'——默认 auto，内存对象已同步

    await session.execute(
        update(ArticleVersion)
        .where(...)
        .values(status="archived")
        .execution_options(synchronize_session=False)
    )
    print(version.status)        # 还是旧值——没人通知它
    await session.refresh(version)   # 要新值：手动对账

比喻：数据库是公告栏，会话里的对象是没看群的同事。公告栏换了通知（UPDATE），同事手里的旧通知单还在。synchronize_session 就是「改公告时顺手在群里 @所有人」；设成 False 就是「只换公告不吭声」——同事继续按旧通知干活，而且不报错。

常见坑

• 不带 where = 全表更新。SQLAlchemy 不会拦（它认为你是故意的），也没有安全开关——养成「先写 where 再写 values」的肌肉记忆。
• synchronize_session=False 后读旧值不报错，是最阴的坑：库里明明改了，日志打出来还是旧的。规则：关了同步就别信内存对象，要新值 refresh()。
• values(title=user_input.upper()) 是 Python 先算好再当绑定参数传；想数据库执行转换，得写 values(title=func.upper(Article.title))——中文场景未必用得上，英文数据清洗很顺手。

12. delete().where()：删除

作用：构造 DELETE 语句，按条件批量删行。

签名：delete(实体).where(条件)。

最小示例：

from sqlalchemy import delete

async with engine.begin() as conn:
    result = await conn.execute(
        delete(ArticleVersion).where(ArticleVersion.status == "draft")
    )
    print(result.rowcount)   # 删掉几行

rowcount == 0 就是「条件没删到东西」——REST 接口里标准的 404 语义，比抛异常再捕获干净：

result = await session.execute(delete(Article).where(Article.id == article_id))
if result.rowcount == 0:
    ...   # 当 404 返回；统一成 HTTPException 是第 09 章的事

常见坑

• 和 update 一样：不带 where = 清空整张表，没有后悔药。
• Core delete 与两套级联（05 章知识的实战验收）：
• ORM 级联 relationship(cascade="all, delete-orphan") 只对 session.delete(obj) 生效——ORM 逐条发 DELETE；
• 数据库级联 ForeignKey(ondelete="CASCADE") 对 Core delete 和任何 SQL 都生效——前提是数据库真的执行外键（PG 默认执行；SQLite 要开 PRAGMA foreign_keys=ON）。

换句话说：用 Core delete 删父行，ORM 级联根本不参与，级不级联全看数据库端的 ondelete。demo_write.py 第 5 节实测：删一篇文章，它的版本行被数据库的 ON DELETE CASCADE 连带清掉。 - 删掉的行若还躺在某个会话的内存里，和 update 同款问题——内存对象不知道自己没了。同样支持 synchronize_session（默认 auto 即可），或者删完别再碰旧对象。

13. SQL 注入的终点站：Core 为什么天然防

第 04 章演示过拼接 SQL 的惨案：攻击串 %' OR 1=1 -- 让全表裸奔。现在从机制上把话说透——Core 不是「更小心地防注入」，而是结构上就没给注入留门。

惨案与解法并排看：

keyword = "' OR 1=1 --"

# ❌ 拼接：数据混进了语句的「结构」——攻击串改变了 SQL 的语法意图
await conn.execute(text(f"SELECT * FROM article WHERE title LIKE '%{keyword}%'"))

# ✅ Core：值只走「参数」通道，结构一个字都没动
stmt = select(Article).where(Article.title.contains(keyword))

编译结果揭示了原理：

print(select(Article).where(Article.channel_id == 1))
# SELECT ... FROM article WHERE article.channel_id = :channel_id_1

你给的值，在构建表达式的那一刻就被装进绑定参数——SQL 骨架（结构）和参数值（数据）从出生起就是两条通道，分开递给数据库。数据库先解析骨架，再把参数当纯文本填进坑里——填什么，都改变不了骨架的形状。攻击串里的引号、OR、注释符，到这里全成了普通字符。

对比一下 text()：它的安全靠自觉（记得写 :占位符，管住 f-string 的手）；Core 的安全靠构造——Article.title == kw 这种写法里，kw 想混进结构都没有入口。想不安全都难，这才叫「天然防」。

最后一个角落：LIKE 的通配符

防注入防的是「数据变成指令」；但 % 和 _ 在 LIKE 里本来就是指令（通配符）。用户在搜索框输入 %，不会被当 SQL 执行，却会让 LIKE 匹配所有行——结构没泄，全量数据泄了，算半个注入：

# autoescape=True：自动转义用户输入里的 % 和 _
# 生成 LIKE '%' || :param || '%' ESCAPE '/'
select(Article).where(Article.title.contains(keyword, autoescape=True))

做搜索框的同学记这一条就够：contains(kw, autoescape=True)。like / ilike / startswith / endswith 全家族都支持 autoescape。

14. 本章速查表

自测三问

1. execute()、session.scalars()、scalar_one() 各拿到什么？后两个在 0 行、多行时分别发生什么？
2. where 和 having 各在什么阶段、对什么对象生效？为什么 having 里能拿 count() 当条件而 where 里不能？
3. 会话里用 synchronize_session=False 执行 update 后，内存对象读到的是什么？两种补救姿势分别是什么？

到这里，「查、改、删」三板斧凑齐，SQL 表达式的用法手册翻完。剩最后一块拼图：表结构本身要演进——加列、改约束、回滚错误决策，create_all 指望不上（05 章的结论），得请带施工日志的正规军 Alembic 了。

下一章：07 · Alembic 数据迁移