第四站练习答案：查询要有预算，写入要有证据

返回纵览 ｜ 专题一 ｜ 专题二

本答案假设使用专题一中的 Ticket、TicketComment 和 open_ticket_cards()。先确认自己的查询合同和目标数据库，再比较代码；答案不会把 SQLite 的方便误报成生产锁的真相。

1｜批量提级：用 F() 避免丢失更新

# tickets/services.py
from django.db import transaction
from django.db.models import F

from .models import Ticket


def refresh_open_ticket_board() -> None:
    # 这里可替换为 cache delete、任务队列投递或 WebSocket 通知。
    print("refresh the open-ticket board")


@transaction.atomic
def promote_open_tickets() -> int:
    changed = (
        Ticket.objects
        .filter(status=Ticket.Status.OPEN, priority__lt=5)
        .update(priority=F("priority") + 1)
    )
    transaction.on_commit(refresh_open_ticket_board)
    return changed

changed 是影响行数；它可以是 0，这不是异常。F() 让加法在数据库内的一条 UPDATE 中完成，两个请求同时提级时不会因为各自在 Python 中读到旧优先级而互相覆盖。update() 不会调用每个实例的 save() 或发送相同的模型信号；若业务规则依赖这些钩子，先重画边界，别在性能优化时悄悄删掉行为。

2｜列表查询：固定预算，而不是固定好运气

# 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(4):
            ticket = Ticket.objects.create(title=f"ticket {index}", assignee=user)
            TicketComment.objects.create(ticket=ticket, body=f"comment {index}")

    def test_card_query_budget_does_not_grow_with_rows(self):
        # Ticket + assignee 一条，comments 预取一条。
        with self.assertNumQueries(2):
            cards = list(open_ticket_cards())
            rendered = [
                (card.title, card.assignee.username, [c.body for c in card.recent_comments])
                for card in cards
            ]

        self.assertEqual(len(rendered), 4)

若列表页不显示评论，删除预取，预算应相应变为 1；若读取的是多值关系，别强行期待 select_related() 创造奇迹。查询数量始终由页面合同决定。

3｜提交后通知：测试的是“注册时机”，不是网络服务

TestCase 会将测试包在事务中并回滚，因此 on_commit() 不会自然执行。用 captureOnCommitCallbacks() 检查注册与执行：

# tickets/tests/test_services.py
from django.test import TestCase

from tickets.models import Ticket
from tickets.services import promote_open_tickets


class PromotionTests(TestCase):
    def test_board_refresh_is_registered_after_successful_update(self):
        Ticket.objects.create(title="tiny", priority=1)

        with self.captureOnCommitCallbacks(execute=True) as callbacks:
            changed = promote_open_tickets()

        self.assertEqual(changed, 1)
        self.assertEqual(len(callbacks), 1)

回滚路径应断言 callback 没有被执行，而不是断言“请求刚好没打出去”。真正的通知系统还要有 outbox、重试和消费者幂等；on_commit() 只保证它不会在数据库事务回滚前抢跑。

4｜两个用户认领同一工单：怎样验证才算数

参考实现有两条正路：

# 条件 UPDATE：规则只有“仍未认领时才领取”。
changed = (
    Ticket.objects
    .filter(pk=ticket_id, status=Ticket.Status.OPEN, assignee__isnull=True)
    .update(status=Ticket.Status.CLAIMED, assignee=user)
)
if changed != 1:
    raise TicketAlreadyClaimed(ticket_id)

# 行锁：要读当前行并依据多个字段/相关操作做决定时，复用专题一的 claim_ticket()。
from django.db import transaction

with transaction.atomic():
    ticket = Ticket.objects.select_for_update().get(pk=ticket_id)
    # 检查不变量，再写入；事务离开时释放锁。

用 TransactionTestCase 和目标数据库做集成测试，或先做专题一的双 dbshell 实验。不要用 SQLite 成功跑过一次就宣称行锁正确：它对 select_for_update() 的锁语义不能提供该证明。测试报告至少应记录数据库厂商、隔离级别、两个连接各自的开始/提交时点，以及最终 assignee_id 只有一个获胜者。

5｜常见错误的判定

写法	为什么不通过
先 get()，Python 里判断 OPEN，再 save()	竞争窗口在读和写之间
在 atomic() 中调用邮件/HTTP，再等待返回	让锁跟着网络延迟一起占用，回滚也撤不回外部动作
吞掉 IntegrityError 后继续查询	事务可能已损坏；把捕获边界移到内层 atomic() 外
prefetch_related() 后仍在循环里再过滤关系	可能重新发查询，或让预取缓存失效

再往前一步

若“工单认领成功”必须可靠地驱动下游系统，建立 outbox 表：和认领在同一数据库事务中写入事件；独立 worker 发送并标记完成；消费者按事件 ID 幂等。即使发送成功但确认写回失败，重复投递也不应造成重复业务效果。这题留作下一层练习，因为它已经跨出 ORM，进入交付系统的边界设计。

继续第五站