QuerySet、关系与事务：让数据库做数据库该做的事

本站纵览 ｜ 练习答案 ｜ 下一篇：执行计划、批量与查询预算

ORM 不是把 SQL 变没了，而是给 SQL 穿上一件 Python 外套。外套好不好看不重要，重要的是你仍知道里面跑了几条查询、锁了哪些行、事务失败时什么会回滚、什么已经飞出去了。

本篇任务

• 用 select_related() 和 prefetch_related() 消除关系读取中的 N+1，而不是盲目预加载整个宇宙。
• 将“认领工单”写成可验证的并发操作：两个请求同时来，只有一个能拿走工单。
• 说清 atomic()、F()、update()、on_commit() 各自保证什么，又不保证什么。
• 在目标数据库上观察 SQL 和执行计划；SQLite 只适合快速开发，不是并发锁行为的替身演员。

一句话心智模型

QuerySet 像一张还没发出去的点菜单：你继续链式调用是在改菜单，迭代、切片、序列化或 list() 才可能让厨房真正开火。transaction.atomic() 则是这桌菜的结账窗口：数据库里的改动要么一起落账，要么一起作废；邮件、缓存、HTTP 请求不在这个收银台里。

可运行起步：一间小型工单室

以下例子适用于 Django 6.0。已有项目可直接创建 tickets app；新项目可按下面命令起步。开发可用 SQLite，但后面涉及 select_for_update() 的实验请切到与生产一致的 PostgreSQL 或 MySQL/InnoDB。

django-admin startproject helpdesk
cd helpdesk
python manage.py startapp tickets
# 把 "tickets" 加入 settings.py 的 INSTALLED_APPS
python manage.py makemigrations tickets
python manage.py migrate
python manage.py shell

1. 模型先把关系说清楚

# tickets/models.py
from django.conf import settings
from django.db import models


class Ticket(models.Model):
    class Status(models.TextChoices):
        OPEN = "open", "Open"
        CLAIMED = "claimed", "Claimed"
        CLOSED = "closed", "Closed"

    title = models.CharField(max_length=200)
    status = models.CharField(max_length=16, choices=Status, default=Status.OPEN)
    priority = models.PositiveSmallIntegerField(default=1)
    assignee = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
        related_name="assigned_tickets",
    )
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)


class TicketComment(models.Model):
    ticket = models.ForeignKey(Ticket, on_delete=models.CASCADE, related_name="comments")
    body = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)

在 shell 中放几条数据：

from django.contrib.auth import get_user_model
from tickets.models import Ticket, TicketComment

User = get_user_model()
ada = User.objects.create_user(username="ada", password="not-for-production")
first = Ticket.objects.create(title="按钮偶尔变成土豆", assignee=ada)
TicketComment.objects.create(ticket=first, body="已复现，土豆很有主见。")
Ticket.objects.create(title="导出 CSV", priority=2)

读路径：先写页面合同，再决定关系装车方式

假设列表页需要工单标题、状态、负责人名字，以及每条工单最近两条评论。这里有两种关系，脾气不同：

关系形状	API	SQL 形状	适合什么
一个 Ticket 对一个 assignee（外键）	select_related("assignee")	在主查询里 JOIN	单值关系，页面一定要读负责人
一个 Ticket 对多个 comments	prefetch_related(...)	主查询后再批量查一次，Python 组装	多值/多对多关系，避免每行一查

# tickets/queries.py
from django.db.models import Count, Prefetch, Q

from .models import Ticket, TicketComment


def open_ticket_cards():
    recent_comments = TicketComment.objects.order_by("-created_at")[:2]

    return (
        Ticket.objects
        .filter(Q(status=Ticket.Status.OPEN) | Q(status=Ticket.Status.CLAIMED))
        .select_related("assignee")
        .prefetch_related(
            Prefetch("comments", queryset=recent_comments, to_attr="recent_comments")
        )
        .only("id", "title", "status", "priority", "assignee__id", "assignee__username")
        .order_by("-priority", "-created_at")
    )


def open_ticket_counts():
    return (
        Ticket.objects
        .values("status")
        .annotate(total=Count("id"))
        .order_by("status")
    )

# 任何 view、serializer 或 shell 中都可以这样消费。
for ticket in open_ticket_cards():
    print(ticket.title, ticket.assignee and ticket.assignee.username)
    print([comment.body for comment in ticket.recent_comments])

print(list(open_ticket_counts()))

为什么不是“一条 SQL 打天下”

select_related() 会把单值关系的列 JOIN 到主查询，通常能在一次查询中取得工单与负责人。prefetch_related() 对多值关系运行额外的批量查询，再把结果贴回对象；它常常是 2 条查询，不是魔法般的 1 条，但不会随着 100 条工单变成 101 条。

另外，only() 是“页面合同”的优化，不是装饰。若模板后来读取了被延迟的 ticket.updated_at，Django 可能为每个对象补一条查询，N+1 会穿着隐身衣回来。优化前先写下页面到底需要哪些字段。

用测试证明查询预算，不靠肉眼数日志

# tickets/tests/test_queries.py
from django.contrib.auth import get_user_model
from django.test import TestCase

from tickets.models import Ticket, TicketComment
from tickets.queries import open_ticket_cards


class TicketQueryTests(TestCase):
    def setUp(self):
        user = get_user_model().objects.create_user(username="ada")
        for index in range(3):
            ticket = Ticket.objects.create(title=f"ticket {index}", assignee=user)
            TicketComment.objects.create(ticket=ticket, body="first")

    def test_cards_have_a_fixed_query_budget(self):
        # 主查询（含 assignee JOIN）+ 评论预取查询。
        with self.assertNumQueries(2):
            rows = list(open_ticket_cards())
            rendered = [
                (row.title, row.assignee.username, [c.body for c in row.recent_comments])
                for row in rows
            ]

        self.assertEqual(len(rendered), 3)

运行 python manage.py test tickets.tests.test_queries。如果有人把 recent_comments 换回 ticket.comments.all()，这个测试会先尖叫，用户不必先等页面变慢才发现事故。

需要看 ORM 实际请求什么时，在开发环境执行：

queryset = open_ticket_cards()
print(queryset.query)              # 审阅 SQL 形状，不要拼接用户输入
print(queryset.explain(analyze=False))  # 目标数据库上看计划；生产慎用 ANALYZE

写路径：为什么“先读再写”会丢工单

这是看起来诚实、并发时却会撒谎的代码：

# 错误示例：两个请求都可能在 status 还是 OPEN 时读到它。
ticket = Ticket.objects.get(pk=ticket_id)
if ticket.status == Ticket.Status.OPEN:
    ticket.status = Ticket.Status.CLAIMED
    ticket.assignee = user
    ticket.save()

时刻	请求 A	请求 B
T1	读到 OPEN
T2		也读到 OPEN
T3	写入 CLAIMED, Ada
T4		写入 CLAIMED, Linus，覆盖 A

数据库不知道“只有一个人能认领”是你的业务不变量，除非你把它表达进条件或锁协议里。

方案 A：需要读—判断—写时，锁住当前行

# tickets/services.py
from functools import partial

from django.db import transaction

from .models import Ticket


class TicketAlreadyClaimed(Exception):
    pass


def notify_ticket_claimed(ticket_id: int) -> None:
    # 真实项目把这里替换为任务队列；它位于提交后，不占用数据库锁。
    print(f"notify: ticket {ticket_id} was claimed")


@transaction.atomic
def claim_ticket(ticket_id: int, user) -> Ticket:
    ticket = (
        Ticket.objects
        .select_for_update()
        .select_related("assignee")
        .get(pk=ticket_id)
    )

    if ticket.status != Ticket.Status.OPEN or ticket.assignee_id is not None:
        raise TicketAlreadyClaimed(ticket_id)

    ticket.status = Ticket.Status.CLAIMED
    ticket.assignee = user
    ticket.save(update_fields=["status", "assignee", "updated_at"])
    transaction.on_commit(partial(notify_ticket_claimed, ticket.pk))
    return ticket

select_for_update() 只有处于事务中才有锁语义。支持行锁的后端在锁冲突时会等待或按照你使用的 nowait / skip_locked 选项处理；SQLite 不会替你演这出戏。在 Django 的 TestCase 里，每个测试本身被包进事务，这还可能掩盖“生产中 autocommit 下会报错”的问题；锁测试通常用 TransactionTestCase 加目标数据库做。

方案 B：只需一个条件写入时，让数据库一次决定

若业务规则只需“仍是未认领才认领”，条件 UPDATE 往往更小、更快，也不需要先把整行搬到 Python：

from django.utils import timezone

updated = (
    Ticket.objects
    .filter(
        pk=ticket_id,
        status=Ticket.Status.OPEN,
        assignee__isnull=True,
    )
    .update(
        status=Ticket.Status.CLAIMED,
        assignee=user,
        updated_at=timezone.now(),
    )
)

if updated != 1:
    raise TicketAlreadyClaimed(ticket_id)

两种方案不互相取代：需要基于多行状态作复杂决定或保留锁到后续写入时，用 select_for_update()；只需原子条件变更时，用条件 UPDATE。update() 会绕过模型的 save() 和相关信号，因此有依赖这些钩子的业务时，别悄悄换成批量写入。

F()：把“加一”交给数据库，而不是读回来再加一

from django.db.models import F

changed = (
    Ticket.objects
    .filter(pk=ticket_id, status=Ticket.Status.OPEN)
    .update(priority=F("priority") + 1)
)

若两个请求都在 Python 里执行 ticket.priority += 1，它们可能读到同一个旧值并覆盖；F("priority") + 1 让加法在数据库的一条 UPDATE 中发生。把 F() 赋给模型实例并 save() 后，记得 refresh_from_db() 再读取真实数值：实例字段可能仍是表达式对象，不是最终整数。

事务边界：atomic() 是收银台，不是时间暂停器

Django 默认处于 autocommit：没有活动事务时，每条查询各自提交。atomic() 把其中的数据库操作组成一个原子块；嵌套 atomic() 通常形成保存点。它不会：

• 撤回已经发送的邮件、已经调用的 HTTP 服务或已经发布的消息；
• 自动恢复 Python 对象、缓存或全局变量里已经修改的值；
• 因为你写了它，就让慢网络请求变快或让锁变得无害。

这就是 on_commit() 的位置：数据库提交成功后再触发缓存失效、邮件或后台任务。

from django.db import IntegrityError, transaction


def create_ticket_with_optional_label(title: str) -> Ticket:
    with transaction.atomic():
        ticket = Ticket.objects.create(title=title)
        try:
            # 内层块像一个小保险箱；唯一约束失败时只回滚这里。
            with transaction.atomic():
                create_optional_unique_label(ticket)
        except IntegrityError:
            record_label_problem(ticket.pk)

        transaction.on_commit(lambda ticket_id=ticket.pk: notify_ticket_created(ticket_id))
        return ticket

不要在同一个 atomic() 块里吞掉 IntegrityError 后继续查询。数据库连接可能已处于需要回滚的状态，Django 会以 TransactionManagementError 提醒你别在破船上继续划。像上例一样把 try/except 放在内层 atomic() 的外面，才保留可用的外层事务。

on_commit() 也不是可靠消息系统：回调只说明本数据库已经提交；它执行失败不会回滚刚才的工单。需要“数据库事件与异步消息最终一致”时，使用 outbox 表、可重试 worker 与消费者幂等，而不是对 on_commit() 许愿。

用两个终端观察真实锁

此实验使用目标数据库的 dbshell，并假定 tickets_ticket 使用支持行锁的事务引擎。先准备一条 OPEN 工单，然后开两个终端。

终端 A：

START TRANSACTION;
SELECT id, status, assignee_id
FROM tickets_ticket
WHERE id = 1
FOR UPDATE;
-- 暂时不要 COMMIT；这行正在被 A 持有。

终端 B：

UPDATE tickets_ticket
SET status = 'claimed'
WHERE id = 1;
-- 在支持行锁的目标数据库上，这里会等待 A 释放锁，或到达锁等待策略。

回到 A 执行 COMMIT;，B 才能继续。实验结束后把数据复原。这个实验验证的是锁，而非“业务接口正确”；业务接口还要测试异常映射、权限、幂等键和提交后通知。

排错路线

1. 结果不对： 确认 QuerySet 是否已被求值；检查过滤条件、排序和 values() 后的结果粒度。
2. 查询太多： 用 assertNumQueries() 固定预算；区分单值关系的 JOIN 和多值关系的预取。
3. 并发仍覆盖： 检查不变量是否写进 filter() 或锁范围；确认在生产同款数据库、同款隔离级别上测试。
4. 事务异常后全都坏了： 把数据库异常的捕获移到嵌套 atomic() 外；不要继续使用已损坏的事务。
5. 通知早到了或根本没到： 检查 on_commit() 是否在事务中注册；进一步判断是否需要 outbox。

版本、兼容性与旧项目提示

• 主线以 Django 6.0 为准。Django 默认 autocommit；ATOMIC_REQUESTS 虽方便，但会把每个 view 包进事务，流式响应、middleware 和模板渲染不都在那个边界里，也可能扩大锁占用。
• select_for_update() 的可用选项、等待行为与锁范围由数据库后端决定。MySQL/InnoDB、PostgreSQL、SQLite 不是可互换的测试替身。
• Django 5.2 LTS 项目可沿用本页主要 ORM 语义；升级前仍需核对项目使用的数据库驱动、迁移和第三方包。

练习：让“批量提级”既快又诚实

产品要把所有 OPEN、优先级低于 5 的工单加 1 分，并在提交成功后通知看板刷新。完成以下任务：

1. 用一条 F() 更新实现批量提级；记录返回的影响行数，并解释为什么不能依靠 Python 循环。
2. 写一个列表查询与 assertNumQueries() 测试：负责人和最新评论不能形成 N+1。
3. 将看板通知放进 on_commit()；分别演示成功提交与回滚时的行为。
4. 设计“两个用户同时认领同一工单”的目标数据库测试或双会话实验，写下你选条件 UPDATE 还是行锁的理由。

完成标准： 主路径、空结果、部分失败/回滚与并发竞争都有可复现证据；你能指出每一条 SQL 是何时执行、每一项外部副作用何时允许发生。

完成后核对第四站答案。答案给实现方向与测试边界，不会替你掩盖目标数据库的真实行为。

官方延伸阅读

• Django 6.0：数据库事务（外部资料已收录于证据账本）
• Django 6.0：QuerySet API（外部资料已收录于证据账本）
• Django 6.0：优化数据库访问（外部资料已收录于证据账本）

课程合同：QuerySet、关联与事务

学习目标

学完本篇，读者应能把「两个并发请求认领同一工单时只有一个成功」拆成输入、可信边界、Django 机制和可复查证据；同时能说明 Django 6.0 与 5.2 LTS 的差异不会改变业务不变量。

心智模型

请求/数据输入 → 边界检查 → Django 核心机制 → 可观察输出 → 恢复动作。配置、数据库提交和外部副作用分别拥有自己的生命周期；没有证据的“应该可以”不算完成。

本地机制图

图中的“输入”是不可信的 HTTP、表单、任务或配置；“边界”负责身份、格式、权限、大小、超时与版本；“机制”才是 URL、View、ORM、事务、缓存、Tasks 或 ASGI；“证据”是测试、响应、日志、查询计划和发布记录。

最小实验

在全新虚拟环境中运行：

python -m django --version
python manage.py check
python manage.py test

实验只验证一个机制：输入一条有效值，再输入一条空值、重复值或无权限值。每次实验都记录 Django 版本、数据库、命令和退出码。

完整场景

把实验接到 tickets 工单流：请求携带用户身份和工单 ID，经过当前专题负责的边界，完成一次读或写，并留下响应状态、数据库变化、日志字段和清理动作。完整场景必须能第二次执行，或明确说明幂等键、事务回滚、缓存失效和任务重试策略。

运行输出

成功路径应出现：

Django 6.0.x
System check identified no issues (0 silenced).
OK

失败路径要保留真实信号，例如 400（输入无效）、403（权限/CSRF）、404（资源不存在）、405（方法不允许）或数据库约束异常；不要把失败改写成含糊的 200。

失败边界

至少验证：空输入、重复提交、匿名用户、越权对象、并发写入、下游不可用、超时和版本不兼容。记录“谁拒绝、何时拒绝、状态是否回滚、重试是否安全”。开发模式的 traceback 不得进入生产响应，缓存不得跨用户复用敏感结果。

原理与排错

先确认版本和 settings，再用 manage.py check 定位启动问题；接着确认 URL/任务入口、输入解析和权限；最后检查 SQL、事务提交、响应和副作用。若结果异常，用最小复现打开 SQL/请求 ID/任务 ID 观测，沿「复现 → 缩小 → 观测 → 修复 → 回归」闭环，不靠重装依赖猜测。

练习（正文不泄答案）

1. 为上述完整场景补一条正常路径和两条失败路径测试，并记录完整输出。
2. 画出本篇机制图，标出身份 owner、数据提交点、缓存/任务副作用和恢复动作。
3. 故意制造一次重复、越权或超时，说明应返回的状态、日志字段和是否可重试。
4. 写一条 Django 5.2 LTS 兼容验收标准，再说明 Django 6.0 的新能力是否需要条件分支。

验收标准：命令可重复、输出可复查、正文没有练习答案；完成后再打开本阶段答案册。

答案册

本阶段答案册提供参考实现、关键推理、失败边界和替代方案；先独立完成练习再对照。

下一篇

下一篇是04-orm-and-transactions/02-query-plans-and-bulk-work，继续沿同一条输入—边界—机制—证据链前进。