Django 后台任务怎么写5 分钟跑通 django-tasks 新手指南【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks你在写订单完成后自动生成 PDF 报表这类功能清楚不能让请求线程干等报表生成完。你需要的是一套后台任务框架而 django-tasks 正是干这个的它是 Django 内置任务系统的向后移植版给你一套统一的定义任务 → 入队 → 拿结果的 API让你不用从第一天就绑定某一种具体队列实现。装完配好第一个任务就能跑第一步装包python -m pip install django-tasks然后在 settings.py 里加两段配置把django_tasks放进INSTALLED_APPS再用TASKS字典指定由哪个后端负责执行任务INSTALLED_APPS [ # ... django_tasks, ] TASKS { default: { BACKEND: django_tasks.backends.immediate.ImmediateBackend } }TASKS这段不写也有默认值就是 ImmediateBackend所以最低成本真的只是装包 加应用。接下来在任意模块比如你的 app 下的 tasks.py里用task()装饰器把一个普通函数包起来它就变成了任务from django_tasks import task task() def calculate_meaning_of_life() - int: return 42 result calculate_meaning_of_life.enqueue()调用enqueue()后任务进入队列返回的TaskResult就是这张快递单号状态、时间戳、返回值全靠它跟踪。任务要收参数时直接把参数透传给enqueue即可位置参数、关键字参数都支持比如send_email.enqueue(ab.com, Hi, 欢迎)不需要任何额外包装。❓ 为什么 enqueue 了任务却没执行新手最常见的困惑就是这个调了enqueue()任务却好像没动。答案的根源在于——任务怎么跑完全由你配置的后端决定内置的两个后端行为截然不同后端行为supports_get_result适合场景ImmediateBackend在当前线程同步执行enqueue()等它跑完才返回否结果不持久化上手体验、必须同步拿结果的场景DummyBackend只记录存下来永不执行是存在内存列表单元测试注意 ImmediateBackend 名字里的 immediate 是立即的意思它其实没有排队enqueue()调用返回时任务已经跑完了。如果你的任务一入队就卡住请求不是框架慢是你选的这个后端就是同步的。想换真正的队列实现只需把TASKS里的BACKEND字符串换掉任务代码一行不动。每个后端还暴露四个supports_*开关让你运行时判断能力边界supports_defer是否支持run_after延迟调度supports_priority是否支持按优先级执行supports_async_task是否支持异步函数supports_get_result事后的另一个请求里能否取回结果建议把先看开关再写代码养成习惯比如if default_task_backend.supports_get_result:不满足就降级处理这样将来换后端业务代码不会翻车。给任务排期run_after 和优先级怎么写task装饰器本身接受几个参数priority-100 到 100 的整数数字越大越优先默认 0、queue_name队列名默认default、backend指定用TASKS里哪个后端、takes_context让任务函数第一个参数收到TaskContext可以据此知道当前是第几次尝试。延迟执行不用改装饰器入队时传run_after就行result my_task.enqueue(run_aftertimezone.now() timedelta(hours1))。前提是后端声明了supports_defer True否则直接抛InvalidTask另外项目开了USE_TZ时run_after必须是带时区的 datetime。想按类别分流任务就用queue_name定义时写task(queue_namespecial)或者入队时临时改——calculate_meaning_of_life.using(priority10)会返回一个改了默认值的副本再enqueue()即可。多队列容易拼错名字可以在TASKS配置里用QUEUES声明允许的名字白名单往未登记的名字入队会被InvalidTask当场拦下。一个 result 对象状态、返回值、错误一次看全enqueue()拿到的TaskResult状态会在四个值之间走READY已入队待跑→ RUNNING执行中→ SUCCESSFUL 或 FAILED。任务函数的返回值落在result.return_value上但要小心对 FAILED 或还没跑完的结果访问return_value会抛ValueError所以先判断result.status TaskResultStatus.SUCCESSFUL再取值。任务挂掉了也不用去翻日志result.errors[0].exception_class给你异常类型.traceback是整段栈信息字符串直接打进你的日志系统即可。如果要在另一个请求、另一台机器上查结果先把result.id存下来当不透明字符串对待之后随时用default_task_backend.get_result(result_id)按 ID 取回任意任务的最新状态而task.get_result(id)只能取同一任务的结果取错了会抛TaskResultMismatch。如果手里的对象是入队瞬间的快照调一下result.refresh()能从存储重新加载。顺便说巡检对于会把结果存下来的后端你可以把状态仍是 RUNNING 且started_at距今超过一小时的挑出来告警字段都是现成的项目仓库 里也展示了各状态的实际流转。 怎么写出不真跑的单元测试测试时把后端切成 DummyBackend 最省事任务只入队不执行你可以放心断言参数和初始状态override_settings( TASKS{default: {BACKEND: django_tasks.backends.dummy.DummyBackend}} ) class MyTaskTests(TestCase): def test_enqueue(self): result send_email.enqueue(ab.com, Hi, 欢迎) self.assertEqual(result.status, TaskResultStatus.READY)此时 status 停在 READY、attempts是 0取return_value也会得到任务还没跑完的报错——这些恰好都是可以断言的行为。DummyBackend把结果堆在内存列表里记得在setUp里调default_task_backend.clear()清掉上一条用例留下的数据写法可以参考 tests 目录 里的现成例子。任务跑完了怎么自动通知我不用自己轮询。django_tasks.signals提供三个标准 Django 信号from django.dispatch import receiver from django_tasks.signals import task_finished receiver(task_finished) def on_task_done(sender, task_result, **kwargs): pass # 在这里接日志、监控、下游通知task_enqueued任务一入队就触发task_started真正开始执行前一刻触发task_finished成功或失败后都触发适合挂告警三者的 sender 都是后端类并通过task_result关键字把对应的结果对象传进来框架自带的日志输出就是用 receiver 实现的细节可以看 django_tasks/signals.py。下一步就做一件事在你的项目里加好INSTALLED_APPS挑一个纯函数用task()包起来enqueue()跑一遍再顺手补一条 DummyBackend 的测试。【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Django 后台任务怎么写：5 分钟跑通 django-tasks 新手指南
Django 后台任务怎么写5 分钟跑通 django-tasks 新手指南【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks你在写订单完成后自动生成 PDF 报表这类功能清楚不能让请求线程干等报表生成完。你需要的是一套后台任务框架而 django-tasks 正是干这个的它是 Django 内置任务系统的向后移植版给你一套统一的定义任务 → 入队 → 拿结果的 API让你不用从第一天就绑定某一种具体队列实现。装完配好第一个任务就能跑第一步装包python -m pip install django-tasks然后在 settings.py 里加两段配置把django_tasks放进INSTALLED_APPS再用TASKS字典指定由哪个后端负责执行任务INSTALLED_APPS [ # ... django_tasks, ] TASKS { default: { BACKEND: django_tasks.backends.immediate.ImmediateBackend } }TASKS这段不写也有默认值就是 ImmediateBackend所以最低成本真的只是装包 加应用。接下来在任意模块比如你的 app 下的 tasks.py里用task()装饰器把一个普通函数包起来它就变成了任务from django_tasks import task task() def calculate_meaning_of_life() - int: return 42 result calculate_meaning_of_life.enqueue()调用enqueue()后任务进入队列返回的TaskResult就是这张快递单号状态、时间戳、返回值全靠它跟踪。任务要收参数时直接把参数透传给enqueue即可位置参数、关键字参数都支持比如send_email.enqueue(ab.com, Hi, 欢迎)不需要任何额外包装。❓ 为什么 enqueue 了任务却没执行新手最常见的困惑就是这个调了enqueue()任务却好像没动。答案的根源在于——任务怎么跑完全由你配置的后端决定内置的两个后端行为截然不同后端行为supports_get_result适合场景ImmediateBackend在当前线程同步执行enqueue()等它跑完才返回否结果不持久化上手体验、必须同步拿结果的场景DummyBackend只记录存下来永不执行是存在内存列表单元测试注意 ImmediateBackend 名字里的 immediate 是立即的意思它其实没有排队enqueue()调用返回时任务已经跑完了。如果你的任务一入队就卡住请求不是框架慢是你选的这个后端就是同步的。想换真正的队列实现只需把TASKS里的BACKEND字符串换掉任务代码一行不动。每个后端还暴露四个supports_*开关让你运行时判断能力边界supports_defer是否支持run_after延迟调度supports_priority是否支持按优先级执行supports_async_task是否支持异步函数supports_get_result事后的另一个请求里能否取回结果建议把先看开关再写代码养成习惯比如if default_task_backend.supports_get_result:不满足就降级处理这样将来换后端业务代码不会翻车。给任务排期run_after 和优先级怎么写task装饰器本身接受几个参数priority-100 到 100 的整数数字越大越优先默认 0、queue_name队列名默认default、backend指定用TASKS里哪个后端、takes_context让任务函数第一个参数收到TaskContext可以据此知道当前是第几次尝试。延迟执行不用改装饰器入队时传run_after就行result my_task.enqueue(run_aftertimezone.now() timedelta(hours1))。前提是后端声明了supports_defer True否则直接抛InvalidTask另外项目开了USE_TZ时run_after必须是带时区的 datetime。想按类别分流任务就用queue_name定义时写task(queue_namespecial)或者入队时临时改——calculate_meaning_of_life.using(priority10)会返回一个改了默认值的副本再enqueue()即可。多队列容易拼错名字可以在TASKS配置里用QUEUES声明允许的名字白名单往未登记的名字入队会被InvalidTask当场拦下。一个 result 对象状态、返回值、错误一次看全enqueue()拿到的TaskResult状态会在四个值之间走READY已入队待跑→ RUNNING执行中→ SUCCESSFUL 或 FAILED。任务函数的返回值落在result.return_value上但要小心对 FAILED 或还没跑完的结果访问return_value会抛ValueError所以先判断result.status TaskResultStatus.SUCCESSFUL再取值。任务挂掉了也不用去翻日志result.errors[0].exception_class给你异常类型.traceback是整段栈信息字符串直接打进你的日志系统即可。如果要在另一个请求、另一台机器上查结果先把result.id存下来当不透明字符串对待之后随时用default_task_backend.get_result(result_id)按 ID 取回任意任务的最新状态而task.get_result(id)只能取同一任务的结果取错了会抛TaskResultMismatch。如果手里的对象是入队瞬间的快照调一下result.refresh()能从存储重新加载。顺便说巡检对于会把结果存下来的后端你可以把状态仍是 RUNNING 且started_at距今超过一小时的挑出来告警字段都是现成的项目仓库 里也展示了各状态的实际流转。 怎么写出不真跑的单元测试测试时把后端切成 DummyBackend 最省事任务只入队不执行你可以放心断言参数和初始状态override_settings( TASKS{default: {BACKEND: django_tasks.backends.dummy.DummyBackend}} ) class MyTaskTests(TestCase): def test_enqueue(self): result send_email.enqueue(ab.com, Hi, 欢迎) self.assertEqual(result.status, TaskResultStatus.READY)此时 status 停在 READY、attempts是 0取return_value也会得到任务还没跑完的报错——这些恰好都是可以断言的行为。DummyBackend把结果堆在内存列表里记得在setUp里调default_task_backend.clear()清掉上一条用例留下的数据写法可以参考 tests 目录 里的现成例子。任务跑完了怎么自动通知我不用自己轮询。django_tasks.signals提供三个标准 Django 信号from django.dispatch import receiver from django_tasks.signals import task_finished receiver(task_finished) def on_task_done(sender, task_result, **kwargs): pass # 在这里接日志、监控、下游通知task_enqueued任务一入队就触发task_started真正开始执行前一刻触发task_finished成功或失败后都触发适合挂告警三者的 sender 都是后端类并通过task_result关键字把对应的结果对象传进来框架自带的日志输出就是用 receiver 实现的细节可以看 django_tasks/signals.py。下一步就做一件事在你的项目里加好INSTALLED_APPS挑一个纯函数用task()包起来enqueue()跑一遍再顺手补一条 DummyBackend 的测试。【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考