Django 任务框架 django-tasks定义、入队与结果回收实战【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasksdjango-tasks 是 Django 5.2 内置任务框架的向后移植版让你在不升级 Django 版本的前提下用统一的task装饰器定义任务、延迟执行并回收结果适合不想立刻引入 Celery 等重型方案的 Django 开发者。1. 适合谁用Django 项目里有报表生成、图片处理、数据同步这类想移出请求主链路的活但暂时不想引入消息队列希望任务 API 与 Django 内置django.tasks保持一致将来升级 Django 后无缝切换需要在测试环境跳过真实执行、只验证任务逻辑的场景不适用需要分布式 Worker 集群、任务自动重试、定时调度的项目——django-tasks 目前只随附两个内置后端立即执行与假执行生产级异步执行需自行接入第三方后端2. 最小可用配置python -m pip install django-tasksINSTALLED_APPS中加入django_tasksINSTALLED_APPS [ # ... django_tasks, ]TASKS设置可省略等价于默认使用ImmediateBackend任务在入队时同步执行TASKS { default: { BACKEND: django_tasks.backends.immediate.ImmediateBackend, }, }BACKEND是后端类路径别名default会被task()和default_task_backend默认引用。3. 一个任务从头到尾定义任务函数并打上task()装饰器以生成月度销售报表为例from django_tasks import task task() def generate_sales_report(month: str) - str: rows fetch_sales(month) return write_report(rows)任务函数必须是模块级函数入队和返回值的参数都应可 JSON 序列化。入队即执行enqueue()直接返回TaskResult对象result generate_sales_report.enqueue(2026-07)TaskResult带有id、status等字段return_value属性只在状态为SUCCESSFUL时返回任务返回值否则抛ValueError。assert result.status TaskResultStatus.SUCCESSFUL path result.return_value # 报表文件路径4. 执行时机立即、延迟与优先级立即执行是默认行为。using()可在不改装饰器的情况下派生新任务覆盖优先级、队列、后端等默认值from datetime import timedelta from django.utils import timezone report_next_month generate_sales_report.using( run_aftertimezone.now() timedelta(days1), priority10, )延迟执行run_after指定最早执行时间USE_TZ开启时必须是带时区的 datetime优先级priority取 -100 到 100 的整数数值越大越先执行两者是否真正生效取决于后端是否支持不支持时定义任务阶段就会抛InvalidTask。5. 结果与失败处理TaskResult.status有四种状态READY已入队、RUNNING、SUCCESSFUL、FAILEDis_finished属性可快速判断是否结束。任务抛异常时不会打断调用方异常信息记在result.errors中每项含exception_class异常类和traceback字符串if result.status TaskResultStatus.FAILED: err result.errors[0] print(err.exception_class, err.traceback)跨请求取回结果分两步入队后保存result.id长度至多 64 字符视同不透明字符串之后用get_result按 ID 查询report_task.get_result(saved_result_id)Task.get_result()只返回属于该任务类型的结果类型不符会抛TaskResultMismatch用default_task_backend.get_result()可查询任意任务的结果。本地缓存的TaskResult可通过refresh()从后端重新加载最新状态。6. 执行环境后端对比与能力探测后端是否执行任务supports_defersupports_get_resultsupports_priority适用场景ImmediateBackend是当前线程同步否否是开发环境、轻量后台工作DummyBackend否仅记录到READY是部分支持是测试运行时用supports_*属性探测后端能力避免在不支持的特性上踩坑from django_tasks import task_backends backend task_backends[default] backend.supports_defer # 是否支持 run_after 延迟 backend.supports_get_result # 是否可跨线程/进程取回结果探测结果还可结合 Django 的 system check 框架在部署阶段校验任务配置是否符合要求。7. 生命周期信号钩子信号触发时机task_enqueued任务入队时task_started任务开始执行前task_finished任务结束成功或失败from django.dispatch import receiver from django_tasks.signals import task_finished receiver(task_finished) def on_task_finished(sender, task_result, **kwargs): ...三个信号的sender都是后端类task_result为对应的TaskResultdjango-tasks 内置了日志接收器会按状态输出 info/exception 日志。8. 测试与生产建议测试时把后端换成DummyBackend任务只入队不执行状态停在READY方便断言入队逻辑而不关心函数体TASKS { default: { BACKEND: django_tasks.backends.dummy.DummyBackend }, }任务函数保持模块级入参和返回值尽量 JSON 可序列化长任务拆小任务内异常会被框架捕获并记入errors不要依赖异常中断进程监控超时任务的前提是结果可查内置后端均不支持supports_get_result需要轮询RUNNING状态时应接入支持该特性的后端9. FAQ为什么enqueue()会让请求卡住默认ImmediateBackend在当前线程同步执行任务执行完才返回。需要不阻塞请求时必须换用支持异步执行的后端。为什么设置了run_after任务还是立刻跑了该后端supports_defer为 False不会等待指定时间只有支持延迟执行的后端才会按run_after调度。为什么结果查不回来内置后端的supports_get_result均为 FalseDummyBackend的结果仅存于当前线程get_result()抛NotImplementedError。跨请求、跨进程取结果需选择支持的后端。10. 速查表功能对应 API / 参数定义任务task(priority, queue_name, backend, takes_context)入队task.enqueue(*args, **kwargs)/await task.aenqueue(...)派生修改默认值task.using(priority, run_after, ...)查状态result.status、result.is_finished、result.attempts取返回值result.return_value、result.refresh()取异常result.errors[0].exception_class/.traceback跨请求取结果task.get_result(result_id)/default_task_backend.get_result(result_id)生命周期钩子task_enqueued/task_started/task_finished能力探测后端实例的supports_defer等四个属性从task()到enqueue()再到TaskResult三步就构成完整闭环。下一步建议先用DummyBackend在测试里跑通入队断言再按业务量评估是否接入支持延迟与结果检索的后端API 细节可参考 README。【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Django 任务框架 django-tasks：定义、入队与结果回收实战
Django 任务框架 django-tasks定义、入队与结果回收实战【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasksdjango-tasks 是 Django 5.2 内置任务框架的向后移植版让你在不升级 Django 版本的前提下用统一的task装饰器定义任务、延迟执行并回收结果适合不想立刻引入 Celery 等重型方案的 Django 开发者。1. 适合谁用Django 项目里有报表生成、图片处理、数据同步这类想移出请求主链路的活但暂时不想引入消息队列希望任务 API 与 Django 内置django.tasks保持一致将来升级 Django 后无缝切换需要在测试环境跳过真实执行、只验证任务逻辑的场景不适用需要分布式 Worker 集群、任务自动重试、定时调度的项目——django-tasks 目前只随附两个内置后端立即执行与假执行生产级异步执行需自行接入第三方后端2. 最小可用配置python -m pip install django-tasksINSTALLED_APPS中加入django_tasksINSTALLED_APPS [ # ... django_tasks, ]TASKS设置可省略等价于默认使用ImmediateBackend任务在入队时同步执行TASKS { default: { BACKEND: django_tasks.backends.immediate.ImmediateBackend, }, }BACKEND是后端类路径别名default会被task()和default_task_backend默认引用。3. 一个任务从头到尾定义任务函数并打上task()装饰器以生成月度销售报表为例from django_tasks import task task() def generate_sales_report(month: str) - str: rows fetch_sales(month) return write_report(rows)任务函数必须是模块级函数入队和返回值的参数都应可 JSON 序列化。入队即执行enqueue()直接返回TaskResult对象result generate_sales_report.enqueue(2026-07)TaskResult带有id、status等字段return_value属性只在状态为SUCCESSFUL时返回任务返回值否则抛ValueError。assert result.status TaskResultStatus.SUCCESSFUL path result.return_value # 报表文件路径4. 执行时机立即、延迟与优先级立即执行是默认行为。using()可在不改装饰器的情况下派生新任务覆盖优先级、队列、后端等默认值from datetime import timedelta from django.utils import timezone report_next_month generate_sales_report.using( run_aftertimezone.now() timedelta(days1), priority10, )延迟执行run_after指定最早执行时间USE_TZ开启时必须是带时区的 datetime优先级priority取 -100 到 100 的整数数值越大越先执行两者是否真正生效取决于后端是否支持不支持时定义任务阶段就会抛InvalidTask。5. 结果与失败处理TaskResult.status有四种状态READY已入队、RUNNING、SUCCESSFUL、FAILEDis_finished属性可快速判断是否结束。任务抛异常时不会打断调用方异常信息记在result.errors中每项含exception_class异常类和traceback字符串if result.status TaskResultStatus.FAILED: err result.errors[0] print(err.exception_class, err.traceback)跨请求取回结果分两步入队后保存result.id长度至多 64 字符视同不透明字符串之后用get_result按 ID 查询report_task.get_result(saved_result_id)Task.get_result()只返回属于该任务类型的结果类型不符会抛TaskResultMismatch用default_task_backend.get_result()可查询任意任务的结果。本地缓存的TaskResult可通过refresh()从后端重新加载最新状态。6. 执行环境后端对比与能力探测后端是否执行任务supports_defersupports_get_resultsupports_priority适用场景ImmediateBackend是当前线程同步否否是开发环境、轻量后台工作DummyBackend否仅记录到READY是部分支持是测试运行时用supports_*属性探测后端能力避免在不支持的特性上踩坑from django_tasks import task_backends backend task_backends[default] backend.supports_defer # 是否支持 run_after 延迟 backend.supports_get_result # 是否可跨线程/进程取回结果探测结果还可结合 Django 的 system check 框架在部署阶段校验任务配置是否符合要求。7. 生命周期信号钩子信号触发时机task_enqueued任务入队时task_started任务开始执行前task_finished任务结束成功或失败from django.dispatch import receiver from django_tasks.signals import task_finished receiver(task_finished) def on_task_finished(sender, task_result, **kwargs): ...三个信号的sender都是后端类task_result为对应的TaskResultdjango-tasks 内置了日志接收器会按状态输出 info/exception 日志。8. 测试与生产建议测试时把后端换成DummyBackend任务只入队不执行状态停在READY方便断言入队逻辑而不关心函数体TASKS { default: { BACKEND: django_tasks.backends.dummy.DummyBackend }, }任务函数保持模块级入参和返回值尽量 JSON 可序列化长任务拆小任务内异常会被框架捕获并记入errors不要依赖异常中断进程监控超时任务的前提是结果可查内置后端均不支持supports_get_result需要轮询RUNNING状态时应接入支持该特性的后端9. FAQ为什么enqueue()会让请求卡住默认ImmediateBackend在当前线程同步执行任务执行完才返回。需要不阻塞请求时必须换用支持异步执行的后端。为什么设置了run_after任务还是立刻跑了该后端supports_defer为 False不会等待指定时间只有支持延迟执行的后端才会按run_after调度。为什么结果查不回来内置后端的supports_get_result均为 FalseDummyBackend的结果仅存于当前线程get_result()抛NotImplementedError。跨请求、跨进程取结果需选择支持的后端。10. 速查表功能对应 API / 参数定义任务task(priority, queue_name, backend, takes_context)入队task.enqueue(*args, **kwargs)/await task.aenqueue(...)派生修改默认值task.using(priority, run_after, ...)查状态result.status、result.is_finished、result.attempts取返回值result.return_value、result.refresh()取异常result.errors[0].exception_class/.traceback跨请求取结果task.get_result(result_id)/default_task_backend.get_result(result_id)生命周期钩子task_enqueued/task_started/task_finished能力探测后端实例的supports_defer等四个属性从task()到enqueue()再到TaskResult三步就构成完整闭环。下一步建议先用DummyBackend在测试里跑通入队断言再按业务量评估是否接入支持延迟与结果检索的后端API 细节可参考 README。【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考