django-tasks 实践教程给 Django 加后台任务队列、延迟任务与结果查询【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks当 Django 项目需要发邮件、生成报表、清理过期数据时同步执行会拖慢响应。django-tasks 让你用一个装饰器把普通函数变成任务通过一次enqueue调用派发并支持延迟执行、状态查询与错误取回。下面从最小配置开始把任务队列接入已有项目。什么时候需要引入 django-tasks几个典型信号请求里存在不影响页面主流程的重活发通知邮件、导出文件、重建缓存希望把某个动作推迟到稍后执行比如一小时后发送提醒发起方需要知道后台操作最终成功与否并能取回返回值或异常。django-tasks 是 Django 官方内置任务框架的向后移植API 与官方方向一致。核心概念只有一个任务后端决定任务是在当前进程执行还是交给外部系统执行。框架自带两个后端不需要额外依赖就能跑起来。让任务队列跑起来安装与最小配置先安装包一条命令即可python -m pip install django-tasks然后在settings.py中启用应用并声明后端。TASKS的写法类似DATABASES支持多个别名后端选择见 后端源码INSTALLED_APPS [ # ... django_tasks, ] TASKS {default: {BACKEND: django_tasks.backends.immediate.ImmediateBackend}}ImmediateBackend在当前线程同步执行任务enqueue返回时任务已经跑完。它适合本地开发和暂时不需要真异步、但想统一接口的阶段。不配置TASKS时框架会回退到同样的默认值但显式写出来更清楚。派发第一个后台任务任务是加了task装饰器的普通函数。函数必须定义在模块顶层不能是闭包或内部函数——后端序列化任务时靠模块路径找回它。以发送欢迎邮件为例这是最常见的真实场景from django_tasks import task task() def send_welcome_email(to: str, subject: str) - None: # 发送邮件的逻辑调用enqueue派发位置参数和关键字参数会原样传给函数返回值是一个TaskResultresult send_welcome_email.enqueue(userexample.com, 欢迎加入) # 派发时按次调整优先级和队列不影响原始定义 send_welcome_email.using(priority10, queue_namereports).enqueue(to, subject)using返回一个新的 Task 实例相当于复制一份并改掉默认值。在异步视图或协程里用await task.aenqueue(...)代替enqueue。控制执行时机与结果配置延迟任务延迟执行通过run_after指定最早执行时间。注意两点项目开启USE_TZ时必须传带时区的datetime而且延迟能力取决于后端——默认的ImmediateBackend不支持run_after直接派发会抛InvalidTask。一小时后再执行写法如下from datetime import timedelta from django.utils import timezone send_welcome_email.using(run_aftertimezone.now() timedelta(hours1)).enqueue(to, subject)派发前可用default_task_backend.supports_defer判断当前后端是否支持配合 Django 的 checks 框架能在启动时暴露配置问题。同理还有supports_priority优先级、supports_async_taskasync 任务函数、supports_get_result跨线程/进程取回结果四个布尔标志。查询任务状态与结果TaskResult的status有四种取值READY已入队未执行、RUNNING、SUCCESSFUL、FAILED。返回值通过return_value属性获取前提是状态为成功否则会抛ValueError失败信息在errors列表里当前实现只会有一个元素if result.status is TaskResultStatus.SUCCESSFUL: value result.return_value elif result.status is TaskResultStatus.FAILED: print(result.errors[0].exception_class) # 异常类型 print(result.errors[0].traceback) # 完整堆栈字符串如果任务在后台被更新过调用result.refresh()从存储重新加载。跨请求取回结果时把result.id当作不透明字符串保存最长 64 字符之后用send_welcome_email.get_result(result_id)取回任务类型不匹配会抛TaskResultMismatch或用default_task_backend.get_result(result_id)取回任意任务——后一种方式只对声明了supports_get_result的后端有意义。选择执行后端与队列划分两个内置后端的分工ImmediateBackendenqueue即执行结果只存在于当前进程内存中重启即失。适合开发和演示。DummyBackend只记录、不执行入队后状态停在READY结果存进后端的results列表当前线程内可见可用clear()清空。它是单元测试的主力。生产环境的分布式执行需要自行实现BaseTaskBackend或选用社区后端django-tasks-db、django-tasks-rq自 0.12 起不再随主包分发。用 QUEUES 约束队列默认所有任务进入default队列。多队列时建议在配置中显式列出允许的名称防止任务派发到没人消费的队列TASKS {default: {BACKEND: ..., QUEUES: [default, reports]}}派发到未知队列会抛InvalidTask把QUEUES设为空列表则关闭校验。指定队列有两个时机定义时task(queue_namereports)固定或派发时.using(queue_namereports)临时切换。任务测试、监控与故障定位用 DummyBackend 验证任务单测中换成DummyBackend既不碰真实邮件等外部资源又能断言任务确实被派发、参数正确。参考仓库里的 测试任务示例override_settings(TASKS{default: {BACKEND: django_tasks.backends.dummy.DummyBackend}}) class WelcomeEmailTests(SimpleTestCase): def test_enqueued(self): result send_welcome_email.enqueue(uexample.com, hi) self.assertEqual(result.status, TaskResultStatus.READY)result.args、result.kwargs记录实际参数result.attempts记录执行次数当前实现只有 0 或 1。失败处理与长任务监控框架提供三个信号task_enqueued、task_started、task_finishedsender是后端类task_result以关键字参数传入。框架已内置接收器会把任务生命周期写入django_tasks日志器失败的日志级别会升级为exception。想接入告警时挂自己的接收器receiver(task_finished) def handle_finished(sender, task_result, **kwargs): if task_result.status is TaskResultStatus.FAILED: error_logger.error(task_result.errors[0].traceback)监控长任务时内置后端没有查询所有运行中任务的接口。务实的做法派发时把result.id记入自己的数据表定期用get_result轮询需后端支持检查started_at、last_attempted_at和is_finished超时未完成的再人工介入。边界在哪django-tasks 不做什么没有取消 API。任务一旦派发无法外部撤回变通方式是在任务函数里检查业务标志比如takes_contextTrue拿到的上下文自行决定中止。结果留存取决于后端。内置两个后端的结果都在内存里重启即失需要长期留存就在任务结束时自行落库。参数与返回值会做 JSON 归一化不可序列化的对象会被转成字符串存储别指望取回原对象。内置后端不做并行。ImmediateBackend的后台任务其实就是同步函数长任务会阻塞请求本身真正的分布式调度、失败自动重试需要换用带 worker 的后端。把重活从请求路径剥离是 django-tasks 解决的第一件事。接口统一后换后端只改TASKS配置任务代码不动。【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 django-tasks 实践教程：给 Django 加后台任务队列、延迟任务与结果查询
django-tasks 实践教程给 Django 加后台任务队列、延迟任务与结果查询【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks当 Django 项目需要发邮件、生成报表、清理过期数据时同步执行会拖慢响应。django-tasks 让你用一个装饰器把普通函数变成任务通过一次enqueue调用派发并支持延迟执行、状态查询与错误取回。下面从最小配置开始把任务队列接入已有项目。什么时候需要引入 django-tasks几个典型信号请求里存在不影响页面主流程的重活发通知邮件、导出文件、重建缓存希望把某个动作推迟到稍后执行比如一小时后发送提醒发起方需要知道后台操作最终成功与否并能取回返回值或异常。django-tasks 是 Django 官方内置任务框架的向后移植API 与官方方向一致。核心概念只有一个任务后端决定任务是在当前进程执行还是交给外部系统执行。框架自带两个后端不需要额外依赖就能跑起来。让任务队列跑起来安装与最小配置先安装包一条命令即可python -m pip install django-tasks然后在settings.py中启用应用并声明后端。TASKS的写法类似DATABASES支持多个别名后端选择见 后端源码INSTALLED_APPS [ # ... django_tasks, ] TASKS {default: {BACKEND: django_tasks.backends.immediate.ImmediateBackend}}ImmediateBackend在当前线程同步执行任务enqueue返回时任务已经跑完。它适合本地开发和暂时不需要真异步、但想统一接口的阶段。不配置TASKS时框架会回退到同样的默认值但显式写出来更清楚。派发第一个后台任务任务是加了task装饰器的普通函数。函数必须定义在模块顶层不能是闭包或内部函数——后端序列化任务时靠模块路径找回它。以发送欢迎邮件为例这是最常见的真实场景from django_tasks import task task() def send_welcome_email(to: str, subject: str) - None: # 发送邮件的逻辑调用enqueue派发位置参数和关键字参数会原样传给函数返回值是一个TaskResultresult send_welcome_email.enqueue(userexample.com, 欢迎加入) # 派发时按次调整优先级和队列不影响原始定义 send_welcome_email.using(priority10, queue_namereports).enqueue(to, subject)using返回一个新的 Task 实例相当于复制一份并改掉默认值。在异步视图或协程里用await task.aenqueue(...)代替enqueue。控制执行时机与结果配置延迟任务延迟执行通过run_after指定最早执行时间。注意两点项目开启USE_TZ时必须传带时区的datetime而且延迟能力取决于后端——默认的ImmediateBackend不支持run_after直接派发会抛InvalidTask。一小时后再执行写法如下from datetime import timedelta from django.utils import timezone send_welcome_email.using(run_aftertimezone.now() timedelta(hours1)).enqueue(to, subject)派发前可用default_task_backend.supports_defer判断当前后端是否支持配合 Django 的 checks 框架能在启动时暴露配置问题。同理还有supports_priority优先级、supports_async_taskasync 任务函数、supports_get_result跨线程/进程取回结果四个布尔标志。查询任务状态与结果TaskResult的status有四种取值READY已入队未执行、RUNNING、SUCCESSFUL、FAILED。返回值通过return_value属性获取前提是状态为成功否则会抛ValueError失败信息在errors列表里当前实现只会有一个元素if result.status is TaskResultStatus.SUCCESSFUL: value result.return_value elif result.status is TaskResultStatus.FAILED: print(result.errors[0].exception_class) # 异常类型 print(result.errors[0].traceback) # 完整堆栈字符串如果任务在后台被更新过调用result.refresh()从存储重新加载。跨请求取回结果时把result.id当作不透明字符串保存最长 64 字符之后用send_welcome_email.get_result(result_id)取回任务类型不匹配会抛TaskResultMismatch或用default_task_backend.get_result(result_id)取回任意任务——后一种方式只对声明了supports_get_result的后端有意义。选择执行后端与队列划分两个内置后端的分工ImmediateBackendenqueue即执行结果只存在于当前进程内存中重启即失。适合开发和演示。DummyBackend只记录、不执行入队后状态停在READY结果存进后端的results列表当前线程内可见可用clear()清空。它是单元测试的主力。生产环境的分布式执行需要自行实现BaseTaskBackend或选用社区后端django-tasks-db、django-tasks-rq自 0.12 起不再随主包分发。用 QUEUES 约束队列默认所有任务进入default队列。多队列时建议在配置中显式列出允许的名称防止任务派发到没人消费的队列TASKS {default: {BACKEND: ..., QUEUES: [default, reports]}}派发到未知队列会抛InvalidTask把QUEUES设为空列表则关闭校验。指定队列有两个时机定义时task(queue_namereports)固定或派发时.using(queue_namereports)临时切换。任务测试、监控与故障定位用 DummyBackend 验证任务单测中换成DummyBackend既不碰真实邮件等外部资源又能断言任务确实被派发、参数正确。参考仓库里的 测试任务示例override_settings(TASKS{default: {BACKEND: django_tasks.backends.dummy.DummyBackend}}) class WelcomeEmailTests(SimpleTestCase): def test_enqueued(self): result send_welcome_email.enqueue(uexample.com, hi) self.assertEqual(result.status, TaskResultStatus.READY)result.args、result.kwargs记录实际参数result.attempts记录执行次数当前实现只有 0 或 1。失败处理与长任务监控框架提供三个信号task_enqueued、task_started、task_finishedsender是后端类task_result以关键字参数传入。框架已内置接收器会把任务生命周期写入django_tasks日志器失败的日志级别会升级为exception。想接入告警时挂自己的接收器receiver(task_finished) def handle_finished(sender, task_result, **kwargs): if task_result.status is TaskResultStatus.FAILED: error_logger.error(task_result.errors[0].traceback)监控长任务时内置后端没有查询所有运行中任务的接口。务实的做法派发时把result.id记入自己的数据表定期用get_result轮询需后端支持检查started_at、last_attempted_at和is_finished超时未完成的再人工介入。边界在哪django-tasks 不做什么没有取消 API。任务一旦派发无法外部撤回变通方式是在任务函数里检查业务标志比如takes_contextTrue拿到的上下文自行决定中止。结果留存取决于后端。内置两个后端的结果都在内存里重启即失需要长期留存就在任务结束时自行落库。参数与返回值会做 JSON 归一化不可序列化的对象会被转成字符串存储别指望取回原对象。内置后端不做并行。ImmediateBackend的后台任务其实就是同步函数长任务会阻塞请求本身真正的分布式调度、失败自动重试需要换用带 worker 的后端。把重活从请求路径剥离是 django-tasks 解决的第一件事。接口统一后换后端只改TASKS配置任务代码不动。【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考