django-tasks 完整指南Django 后台任务框架从入门到落地【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks在 Web 请求里同步跑一个耗时任务页面就会卡住——这是每个 Django 开发者都踩过的坑。django-tasks 是 Django 内置任务框架的向后移植版本你把函数用task装饰一下就能完成任务入队、延迟执行、状态跟踪而不用一上来就引入 Celery。一分钟装好 django-tasks整个配置分三步五分钟内能跑通。第一步安装python -m pip install django-tasks第二步注册 appINSTALLED_APPS [ # ... django_tasks, ]第三步配置后端。后端决定任务由谁执行。不写TASKS时默认就是 ImmediateBackendTASKS { default: { # 当前线程立即执行开发期够用 BACKEND: django_tasks.backends.immediate.ImmediateBackend, } }⚙️ 一个细节任务函数必须定义在模块顶层闭包或嵌套函数会被拒绝。这是框架在装饰时就校验的早暴露早好。定义第一个任务从 task 到任务入队装饰器有两种写法from django_tasks import task task() def calculate_meaning_of_life() - int: return 42 task # 裸装饰器等价写法 def noop_task() - None: return None装饰器接受几个关键字参数priority优先级-100 到 100数字越大越先执行默认 0queue_name任务所属的任务队列默认defaultbackend使用TASKS中哪个后端别名takes_context是否向任务注入上下文后面细讲带参数入队enqueue()的入参会原样透传给任务函数位置参数、关键字参数都行task() def send_invoice(order_id: int, cc: bool False) - None: ... result send_invoice.enqueue(42, ccTrue) # 参数直接透传延迟执行run_after想一小時后执行通过.using()派生一个新 Task 再入队注意不是往enqueue里塞参数from datetime import timedelta from django.utils import timezone later calculate_meaning_of_life.using( run_aftertimezone.now() timedelta(hours1) # 最早执行时间 ) result later.enqueue()⚠️ 前提是后端声明了supports_defer。默认的 ImmediateBackend 不支持指定run_after会直接抛InvalidTaskDummyBackend 支持。生产环境请换真正的队列后端。任务结果怎么查TaskResult 与状态跟踪enqueue()返回一个TaskResult任务的完整生命周期都挂在它上面。判断状态、取返回值状态是四值枚举TaskResultStatusREADY→RUNNING→SUCCESSFUL/FAILED。from django_tasks import TaskResultStatus if result.status TaskResultStatus.SUCCESSFUL: print(result.return_value) # 取返回值非 SUCCESSFUL 会抛 ValueError两个辅助属性很实用result.is_finished是否已出结果和result.attempts执行次数。如果任务是在别处跑的本地这份对象可能过期调result.refresh()可以从后端重新拉取最新状态。跨位置检索结果把result.id存进数据库或传给前端之后在任意位置都能找回result_id result.id # 稍后在另一个请求里 r1 send_invoice.get_result(result_id) # 限定任务类型类型不符会抛异常 r2 default_task_backend.get_result(result_id) # 不关心任务类型id 是不透明的字符串最长 64 字符生成规则由后端决定别自己去解析。读取失败信息任务抛异常不会炸到你的进程里错误被收进errors列表if result.status TaskResultStatus.FAILED: err result.errors[0] print(err.exception_class) # 异常类本身 print(err.traceback) # 完整堆栈是字符串traceback 被压成了字符串排障时打出来看即可。当前版本errors实际只会有单个元素。任务后端与任务队列怎么配 后端是 django-tasks 的核心抽象内置两个后端行为延迟执行异步任务优先级典型用途ImmediateBackend当前线程立即执行❌✅✅开发、默认DummyBackend只存储、不执行✅✅✅测试两者都不设置supports_get_result——也就是说结果只能就地拿不保证跨进程可查。需要 Redis/RQ 之类的持久化后端社区有对应扩展包可以接入。特性检测后端能力不同写兼容代码前先问一句from django_tasks import default_task_backend if default_task_backend.supports_priority: # 只有支持的后端才允许设置非零 priority否则装饰时就报错 t my_task.using(priority10)可用的开关就四个supports_defer、supports_async_task、supports_get_result、supports_priority。队列名怎么管用QUEUES声明合法队列名防止任务被塞进没人消费的任务队列TASKS { default: { BACKEND: django_tasks.backends.immediate.ImmediateBackend, QUEUES: [default, special], # 白名单 } }指定队列有两个时机定义期task(queue_namespecial)使用时my_task.using(queue_namebackground).enqueue()入队到白名单之外的队列会抛InvalidTask。把QUEUES设成[]可关掉校验。任务上下文、信号与可观测性takes_context任务知道自己是第几次执行from django_tasks import task, TaskContext task(takes_contextTrue) def retryable_job(context: TaskContext, order_id: int) - None: print(f第 {context.attempt} 次尝试, order{order_id})开启后第一个参数必须叫context框架会校验。TaskContext目前暴露两个属性task_result当前任务结果和attempt执行次数。生命周期信号三个信号覆盖任务全周期sender是后端类任务结果通过task_result关键字传入from django.dispatch import receiver from django_tasks.signals import task_finished receiver(task_finished) def on_finished(sender, task_result, **kwargs): if task_result.status TaskResultStatus.FAILED: alert(task_result.id) # 接上你的告警渠道task_enqueued任务入队时task_started开始执行前task_finished执行结束成功或失败轮询超时任务框架不会主动告诉你任务卡住了自己设个截止时间from datetime import timedelta from django.utils import timezone deadline timezone.now() timedelta(seconds60) while not result.is_finished and timezone.now() deadline: result.refresh() # 从后端重新拉取状态 time.sleep(1) if result.status TaskResultStatus.RUNNING: log.warning(task %s 疑似超时, result.id) 配合task_started信号把任务 id 记进数据库就能定期扫出长时间停在RUNNING的任务。上生产前的避坑清单 每条都是真实会踩的坑按问题 → 结论说任务函数放哪→ 必须在模块顶层嵌套函数装饰时即抛InvalidTask。ImmediateBackend 能扛生产吗→ 不能。它在当前线程同步执行任务不返回、请求不返回。生产请接真正的队列后端。测试怎么写→ 切到DummyBackendenqueue后断言status READY、args/kwargs是否符合预期副作用为零用完可调backend.clear()清空。失败怎么兜底→KeyboardInterrupt会让任务真正中断进程别在任务里吞它一般异常看errors并用logger.exception接入监控。任务能取消吗→ 框架不提供 cancel API。需要在任务体内自查标志位或选支持取消的后端。结果保存多久→ 取决于后端。内存型后端重启即失重要结果请在task_finished信号里落库。run_after 传本地时间会怎样→USE_TZ True时传 naive datetime 直接抛InvalidTask务必用timezone.now()生成。写在最后✅ 如果你的 Django 项目需要把邮件、报表、第三方回调这类 django 异步任务从请求链路里拆出去又不想引入 Celery 的部署复杂度django-tasks 是现成的任务队列抽象接口对齐 Django 官方方向后端可替换迁移成本很低。现在就挑一个你最想挪出请求的耗时函数给它贴上task()跑一遍——你会发现剩下的都顺理成章。【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
django-tasks 完整指南Django 后台任务框架从入门到落地【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks在 Web 请求里同步跑一个耗时任务页面就会卡住——这是每个 Django 开发者都踩过的坑。django-tasks 是 Django 内置任务框架的向后移植版本你把函数用task装饰一下就能完成任务入队、延迟执行、状态跟踪而不用一上来就引入 Celery。一分钟装好 django-tasks整个配置分三步五分钟内能跑通。第一步安装python -m pip install django-tasks第二步注册 appINSTALLED_APPS [ # ... django_tasks, ]第三步配置后端。后端决定任务由谁执行。不写TASKS时默认就是 ImmediateBackendTASKS { default: { # 当前线程立即执行开发期够用 BACKEND: django_tasks.backends.immediate.ImmediateBackend, } }⚙️ 一个细节任务函数必须定义在模块顶层闭包或嵌套函数会被拒绝。这是框架在装饰时就校验的早暴露早好。定义第一个任务从 task 到任务入队装饰器有两种写法from django_tasks import task task() def calculate_meaning_of_life() - int: return 42 task # 裸装饰器等价写法 def noop_task() - None: return None装饰器接受几个关键字参数priority优先级-100 到 100数字越大越先执行默认 0queue_name任务所属的任务队列默认defaultbackend使用TASKS中哪个后端别名takes_context是否向任务注入上下文后面细讲带参数入队enqueue()的入参会原样透传给任务函数位置参数、关键字参数都行task() def send_invoice(order_id: int, cc: bool False) - None: ... result send_invoice.enqueue(42, ccTrue) # 参数直接透传延迟执行run_after想一小時后执行通过.using()派生一个新 Task 再入队注意不是往enqueue里塞参数from datetime import timedelta from django.utils import timezone later calculate_meaning_of_life.using( run_aftertimezone.now() timedelta(hours1) # 最早执行时间 ) result later.enqueue()⚠️ 前提是后端声明了supports_defer。默认的 ImmediateBackend 不支持指定run_after会直接抛InvalidTaskDummyBackend 支持。生产环境请换真正的队列后端。任务结果怎么查TaskResult 与状态跟踪enqueue()返回一个TaskResult任务的完整生命周期都挂在它上面。判断状态、取返回值状态是四值枚举TaskResultStatusREADY→RUNNING→SUCCESSFUL/FAILED。from django_tasks import TaskResultStatus if result.status TaskResultStatus.SUCCESSFUL: print(result.return_value) # 取返回值非 SUCCESSFUL 会抛 ValueError两个辅助属性很实用result.is_finished是否已出结果和result.attempts执行次数。如果任务是在别处跑的本地这份对象可能过期调result.refresh()可以从后端重新拉取最新状态。跨位置检索结果把result.id存进数据库或传给前端之后在任意位置都能找回result_id result.id # 稍后在另一个请求里 r1 send_invoice.get_result(result_id) # 限定任务类型类型不符会抛异常 r2 default_task_backend.get_result(result_id) # 不关心任务类型id 是不透明的字符串最长 64 字符生成规则由后端决定别自己去解析。读取失败信息任务抛异常不会炸到你的进程里错误被收进errors列表if result.status TaskResultStatus.FAILED: err result.errors[0] print(err.exception_class) # 异常类本身 print(err.traceback) # 完整堆栈是字符串traceback 被压成了字符串排障时打出来看即可。当前版本errors实际只会有单个元素。任务后端与任务队列怎么配 后端是 django-tasks 的核心抽象内置两个后端行为延迟执行异步任务优先级典型用途ImmediateBackend当前线程立即执行❌✅✅开发、默认DummyBackend只存储、不执行✅✅✅测试两者都不设置supports_get_result——也就是说结果只能就地拿不保证跨进程可查。需要 Redis/RQ 之类的持久化后端社区有对应扩展包可以接入。特性检测后端能力不同写兼容代码前先问一句from django_tasks import default_task_backend if default_task_backend.supports_priority: # 只有支持的后端才允许设置非零 priority否则装饰时就报错 t my_task.using(priority10)可用的开关就四个supports_defer、supports_async_task、supports_get_result、supports_priority。队列名怎么管用QUEUES声明合法队列名防止任务被塞进没人消费的任务队列TASKS { default: { BACKEND: django_tasks.backends.immediate.ImmediateBackend, QUEUES: [default, special], # 白名单 } }指定队列有两个时机定义期task(queue_namespecial)使用时my_task.using(queue_namebackground).enqueue()入队到白名单之外的队列会抛InvalidTask。把QUEUES设成[]可关掉校验。任务上下文、信号与可观测性takes_context任务知道自己是第几次执行from django_tasks import task, TaskContext task(takes_contextTrue) def retryable_job(context: TaskContext, order_id: int) - None: print(f第 {context.attempt} 次尝试, order{order_id})开启后第一个参数必须叫context框架会校验。TaskContext目前暴露两个属性task_result当前任务结果和attempt执行次数。生命周期信号三个信号覆盖任务全周期sender是后端类任务结果通过task_result关键字传入from django.dispatch import receiver from django_tasks.signals import task_finished receiver(task_finished) def on_finished(sender, task_result, **kwargs): if task_result.status TaskResultStatus.FAILED: alert(task_result.id) # 接上你的告警渠道task_enqueued任务入队时task_started开始执行前task_finished执行结束成功或失败轮询超时任务框架不会主动告诉你任务卡住了自己设个截止时间from datetime import timedelta from django.utils import timezone deadline timezone.now() timedelta(seconds60) while not result.is_finished and timezone.now() deadline: result.refresh() # 从后端重新拉取状态 time.sleep(1) if result.status TaskResultStatus.RUNNING: log.warning(task %s 疑似超时, result.id) 配合task_started信号把任务 id 记进数据库就能定期扫出长时间停在RUNNING的任务。上生产前的避坑清单 每条都是真实会踩的坑按问题 → 结论说任务函数放哪→ 必须在模块顶层嵌套函数装饰时即抛InvalidTask。ImmediateBackend 能扛生产吗→ 不能。它在当前线程同步执行任务不返回、请求不返回。生产请接真正的队列后端。测试怎么写→ 切到DummyBackendenqueue后断言status READY、args/kwargs是否符合预期副作用为零用完可调backend.clear()清空。失败怎么兜底→KeyboardInterrupt会让任务真正中断进程别在任务里吞它一般异常看errors并用logger.exception接入监控。任务能取消吗→ 框架不提供 cancel API。需要在任务体内自查标志位或选支持取消的后端。结果保存多久→ 取决于后端。内存型后端重启即失重要结果请在task_finished信号里落库。run_after 传本地时间会怎样→USE_TZ True时传 naive datetime 直接抛InvalidTask务必用timezone.now()生成。写在最后✅ 如果你的 Django 项目需要把邮件、报表、第三方回调这类 django 异步任务从请求链路里拆出去又不想引入 Celery 的部署复杂度django-tasks 是现成的任务队列抽象接口对齐 Django 官方方向后端可替换迁移成本很低。现在就挑一个你最想挪出请求的耗时函数给它贴上task()跑一遍——你会发现剩下的都顺理成章。【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
