异步请求与 Django Tasks
异步视图
# tickets/async_views.py
from django.http import JsonResponse
from .models import Ticket
async def open_ticket_count(request):
total = await Ticket.objects.filter(status=Ticket.Status.OPEN).acount()
return JsonResponse({"open_tickets": total})# tickets/tests.py
from django.test import AsyncClient, TestCase
class AsyncStatusTests(TestCase):
async def test_open_count(self):
response = await AsyncClient().get("/async/open-ticket-count/")
self.assertEqual(response.status_code, 200)在 WSGI 下 async view 可以运行，但每请求有适配成本，也得不到完整异步栈的长连接优势。同步 middleware 会导致请求栈适配；启动时检查 Django 日志。
Django 6.0 Tasks
# tickets/tasks.py
from django.core.mail import send_mail
from django.tasks import task
@task(queue_name="emails")
def send_ticket_notice(ticket_id, recipient):
return send_mail(
subject=f"Ticket {ticket_id} changed",
message="Open the ticket desk for details.",
from_email=None,
recipient_list=[recipient],
)from functools import partial
from django.db import transaction
from .tasks import send_ticket_notice
transaction.on_commit(
partial(send_ticket_notice.enqueue, ticket_id=ticket.pk, recipient=user.email)
)开发默认 ImmediateBackend 会立即执行，不代表后台运行。DummyBackend 可记录入队结果但不执行，适合合同测试。
深水边界
- async 不会把阻塞 SDK 变成非阻塞；识别数据库驱动、文件、模板、邮件和第三方调用。
- Task 参数/返回值必须经过 JSON 往返；不要传模型、datetime 或数据库连接。
- 入队成功不是执行成功；设置重试上限、超时、死信/失败检查和告警。
- 至少一次执行要求幂等。使用业务唯一键、状态机或唯一约束。
- Django 本体不提供 WebSocket 服务模型；真实双向实时通常需要 Channels/独立服务并单独核对兼容性。
验收
异步视图由 AsyncClient 通过；Task 参数可以 JSON 往返；DummyBackend 测试能观察入队；重复执行不重复改变业务结果。
练习（先独立完成）
- 将测试 backend 改为 DummyBackend，断言只入队一次。
- 给任务增加业务幂等记录，模拟重复调用。
- 找出项目中的同步 middleware 和阻塞库，写异步采用清单。
深度加餐：异步 view 不会把同步 ORM 变成魔法
async 只改变等待方式；同步数据库调用仍会阻塞事件循环，必须明确跨边界。
先做一个可观察实验
from asgiref.sync import sync_to_async
from django.http import JsonResponse
@sync_to_async
def count_open_tickets():
return Ticket.objects.filter(status='OPEN').count()
async def open_count(request):
count = await count_open_tickets()
return JsonResponse({'count': count})预期：同步 ORM 在线程边界执行，事件循环可以继续处理其他连接；不要在 async view 里直接调用 QuerySet 求值。
原理下潜与工程边界
ASGI server 管理事件循环与连接，Django 通过 sync_to_async/sync_to_async(thread_sensitive=True) 承接同步代码。Django 6.0 Tasks 提供任务接口，但执行器、重试、持久化仍是部署责任。
事务连接不能跨任意线程共享；懒惰 QuerySet 可能在 await 之后才求值；CPU 密集工作应进进程/任务队列，不要塞进 async view。
反例排错路线
记录 event-loop 阻塞、线程池使用和数据库连接数；先把 ORM 封装成同步函数并在边界 await，再逐步测量而不是全局 async 化。
练习加码（仍不提供答案）
加码：把批量报表拆成后台任务，API 只返回 task id，并设计查询状态、取消与超时合同。
验收时请补充：正常路径、权限或失败路径、以及一条能证明事务/资源/配置边界没有被绕过的测试。
继续查阅：Django 官方文档（外部资料已收录于证据账本） · Django 安全清单（外部资料已收录于证据账本）
课程合同：异步请求与 Django Tasks
学习目标
学完本篇，读者应能把「async view 不阻塞事件循环，任务由 worker 边界消费」拆成输入、可信边界、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 观测，沿「复现 → 缩小 → 观测 → 修复 → 回归」闭环，不靠重装依赖猜测。
练习（正文不泄答案）
- 为上述完整场景补一条正常路径和两条失败路径测试，并记录完整输出。
- 画出本篇机制图，标出身份 owner、数据提交点、缓存/任务副作用和恢复动作。
- 故意制造一次重复、越权或超时，说明应返回的状态、日志字段和是否可重试。
- 写一条 Django 5.2 LTS 兼容验收标准，再说明 Django 6.0 的新能力是否需要条件分支。
验收标准：命令可重复、输出可复查、正文没有练习答案；完成后再打开本阶段答案册。
答案册
本阶段答案册提供参考实现、关键推理、失败边界和替代方案；先独立完成练习再对照。
下一篇
下一篇是08-async-tasks-realtime/02-task-idempotency-asgi-lifecycle，继续沿同一条输入—边界—机制—证据链前进。