django-tasks 配置清单5 分钟让 Django 异步任务从零到生产【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasksdjango-tasks 是 Django 官方任务框架的向后移植在 Django 5.2 上用官方的 task 装饰器、enqueue 入队与结果追踪处理 Django 异步任务、定时任务与任务队列无需引入 Celery。为什么是任务框架而不是 Celery电商后台里注册流程同步发验证邮件商品页同步压缩图片订单超时清理也搭在用户请求里执行——接口响应从几十毫秒涨到几秒。这类慢操作有两个出口要么引入 Celery 加一整套 broker 运维要么用 Django 官方自带的任务框架。django-tasks 就是官方框架的向后移植让没到该版本的 Django 项目直接用上这套原生任务系统。写法和官方文档一致以后升级版本只需把导入路径换成django.tasks业务代码原样保留。5 分钟配置清单跑通第一个任务装包、注册、配后端三步走完就可以发出第一个任务。python -m pip install django-tasks下面这段在settings.py里完成注册与后端选择INSTALLED_APPS [ # ... django_tasks, ] TASKS { default: { BACKEND: django_tasks.backends.immediate.ImmediateBackend, }, }配置不写也不报错——默认值就是 ImmediateBackend。各配置项的作用如下配置项作用INSTALLED_APPS中的django_tasks注册应用加载信号与系统检查TASKS按别名组织后端代码里以别名引用BACKEND后端类路径决定任务在哪儿、怎么执行QUEUES该后端允许的队列名白名单默认只有default然后在任务模块里定义并执行第一个任务from django_tasks import task task() def compress_product_image(image_id: int) - str: # 压缩商品图返回产物路径 ... result compress_product_image.enqueue(1024) print(result.status) # TaskResultStatus.SUCCESSFULenqueue()的返回值TaskResult是后续一切状态追踪的入口下一节展开。task 装饰器的四种写法任务本质是一个模块级函数 一组执行参数装饰器参数决定它去哪儿跑、以什么顺序跑。不带参数最简单等价于默认值task def send_verification_email(address: str) - None: ...带参数形式覆盖优先级、队列与后端task(priority10, queue_namemedia, backenddefault) def compress_product_image(image_id: int) - str: ...不想改定义时用using()生成一份覆盖默认值的副本原任务不受影响urgent compress_product_image.using(priority80, queue_namemedia) urgent.enqueue(2048)带上下文时函数第一个参数必须命名为contextfrom django_tasks import task, TaskContext task(takes_contextTrue) def generate_weekly_report(context: TaskContext, year: int, week: int) - str: print(fattempt{context.attempt}) ...参数默认说明priority0-100 到 100 的整数越大越先执行queue_namedefault入队所用队列名backenddefaultTASKS里的后端别名takes_contextFalse函数首参context收到TaskContext两个约束任务函数必须定义在模块级嵌套函数会在装饰器阶段直接抛InvalidTaskcontext提供task_result当前TaskResult与attempt尝试次数需要任务内部自报状态时用。执行与结果怎么查enqueue()把任务交给后端Immediate 后端当场执行完才返回真正的队列后端则先落队列、稍后执行此时拿到的是READY状态的结果。result send_verification_email.enqueue(userexample.com) # 指定时刻执行后端不支持延迟时会被拒绝 deferred generate_weekly_report.using(run_afterplanned_time).enqueue(2026, 34)状态、返回值与错误都挂在TaskResult上from django_tasks import TaskResultStatus if result.status TaskResultStatus.SUCCESSFUL: print(result.return_value) elif result.status TaskResultStatus.FAILED: print(result.errors[0].exception_class) print(result.errors[0].traceback)四个状态是READY入队待执行、RUNNING执行中、SUCCESSFUL、FAILED。注意return_value只在SUCCESSFUL时可用失败或未完成时访问会抛ValueError。跨请求或跨任务要拿结果先存result.id之后按 id 取回result_id result.id # 长度不超过 64 的字符串存库即可 # 任务侧取回id 与任务不匹配会抛 TaskResultMismatch later generate_weekly_report.get_result(result_id) # 后端侧取回任意任务的结果 from django_tasks import default_task_backend any_result default_task_backend.get_result(result_id)结果对象是快照后台状态变了就调result.refresh()刷新is_finished可快速判断是否已终态。监控卡在RUNNING的长任务时框架没有现成的批量查询接口稳妥做法是自己落一份 id 台账周期任务里逐个get_result再按started_at判断是否超时for rid in pending_ids: r default_task_backend.get_result(rid) if r.status TaskResultStatus.RUNNING and r.started_at now - one_hour: flag_stuck(r)后端与队列怎么搭配两个内置后端解决的是两个不同问题一个让功能先跑起来一个让测试不依赖执行。ImmediateBackend默认DummyBackend测试用行为当前线程立即执行只记录TaskResult不执行函数体执行位置请求线程内无执行supports_defer否是supports_async_task是是supports_get_result否结果不跨请求保留是但结果只在当前线程可见适用开发期、无队列时的兜底测试入队行为与参数用supports_*标志在运行时确认能力避免部署后才发现功能不可用from django_tasks import default_task_backend if default_task_backend.supports_defer: scheduled generate_weekly_report.using(run_afterplanned_time)四个标志的含义supports_defer能否run_after延迟执行、supports_async_task能否入队协程、supports_get_result能否从任意线程/进程按 id 取结果、supports_priority能否按优先级排序执行。⚠️ 两个内置后端都覆盖不了跨请求真异步Immediate 在请求线程里同步跑慢任务照样拖长响应。生产环境需要搭配真正基于队列的后端db 或 RQ 系扩展导入路径见django_tasks/backends/base.py的BaseTaskBackend选型时先查这四个标志。队列名是逻辑分组QUEUES写白名单后未列出的队列名在装饰或入队时抛InvalidTask防止打错字导致任务消失设为[]则关闭校验。典型分法TASKS { default: { BACKEND: django_tasks.backends.immediate.ImmediateBackend, QUEUES: [default, media, reports], }, }IO 密集任务压缩图片、发邮件放media计算任务周报放reports慢的挤不占快的。用信号监听任务生命周期三个信号对应任务的完整生命周期task_enqueued入队后、task_started执行前一刻、task_finished执行结束无论成败。handler 以关键字参数收到task_resultfrom django.dispatch import receiver from django_tasks import TaskResultStatus from django_tasks.signals import task_finished receiver(task_finished) def on_task_finished(sender, task_result, **kwargs): if task_result.status TaskResultStatus.FAILED: alert(task_result.task.module_path, task_result.errors[0].traceback)sender是后端类可以据此区分不同后端框架自身已内置了打日志的 handler失败任务会走logger.exception带出 traceback。埋点、通知、统计都挂在这里不用改业务代码。上生产前的避坑清单最后对照这十条过一遍能少掉进大多数坑任务函数定义在模块级嵌套函数抛InvalidTaskrun_after仅对supports_defer的后端有效否则任务立即执行结果保留时长取决于后端Immediate 的结果随进程结束消失长保留需换支持持久化的后端没有取消 API已入队任务撤不回靠任务内部检查取消标记实现业务级取消查失败任务看errors[0].exception_class与errors[0].tracebackreturn_value在FAILED/未完成时抛ValueError访问前先判statusQUEUES是白名单而非注册表写空列表[]才关闭校验priority只能是 -100 到 100 的整数后端不支持时设非零值直接抛错长任务自己落 id 台账 周期get_result做超时巡检attempts当前取值只有 0 或 1不要拿它当重试计数依赖官方任务框架的 API 边界很克制定义、入队、追踪、信号四件事闭环。把 django-tasks 当底座慢操作移出请求链路剩下的交给队列。【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 django-tasks 配置清单：5 分钟让 Django 异步任务从零到生产
django-tasks 配置清单5 分钟让 Django 异步任务从零到生产【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasksdjango-tasks 是 Django 官方任务框架的向后移植在 Django 5.2 上用官方的 task 装饰器、enqueue 入队与结果追踪处理 Django 异步任务、定时任务与任务队列无需引入 Celery。为什么是任务框架而不是 Celery电商后台里注册流程同步发验证邮件商品页同步压缩图片订单超时清理也搭在用户请求里执行——接口响应从几十毫秒涨到几秒。这类慢操作有两个出口要么引入 Celery 加一整套 broker 运维要么用 Django 官方自带的任务框架。django-tasks 就是官方框架的向后移植让没到该版本的 Django 项目直接用上这套原生任务系统。写法和官方文档一致以后升级版本只需把导入路径换成django.tasks业务代码原样保留。5 分钟配置清单跑通第一个任务装包、注册、配后端三步走完就可以发出第一个任务。python -m pip install django-tasks下面这段在settings.py里完成注册与后端选择INSTALLED_APPS [ # ... django_tasks, ] TASKS { default: { BACKEND: django_tasks.backends.immediate.ImmediateBackend, }, }配置不写也不报错——默认值就是 ImmediateBackend。各配置项的作用如下配置项作用INSTALLED_APPS中的django_tasks注册应用加载信号与系统检查TASKS按别名组织后端代码里以别名引用BACKEND后端类路径决定任务在哪儿、怎么执行QUEUES该后端允许的队列名白名单默认只有default然后在任务模块里定义并执行第一个任务from django_tasks import task task() def compress_product_image(image_id: int) - str: # 压缩商品图返回产物路径 ... result compress_product_image.enqueue(1024) print(result.status) # TaskResultStatus.SUCCESSFULenqueue()的返回值TaskResult是后续一切状态追踪的入口下一节展开。task 装饰器的四种写法任务本质是一个模块级函数 一组执行参数装饰器参数决定它去哪儿跑、以什么顺序跑。不带参数最简单等价于默认值task def send_verification_email(address: str) - None: ...带参数形式覆盖优先级、队列与后端task(priority10, queue_namemedia, backenddefault) def compress_product_image(image_id: int) - str: ...不想改定义时用using()生成一份覆盖默认值的副本原任务不受影响urgent compress_product_image.using(priority80, queue_namemedia) urgent.enqueue(2048)带上下文时函数第一个参数必须命名为contextfrom django_tasks import task, TaskContext task(takes_contextTrue) def generate_weekly_report(context: TaskContext, year: int, week: int) - str: print(fattempt{context.attempt}) ...参数默认说明priority0-100 到 100 的整数越大越先执行queue_namedefault入队所用队列名backenddefaultTASKS里的后端别名takes_contextFalse函数首参context收到TaskContext两个约束任务函数必须定义在模块级嵌套函数会在装饰器阶段直接抛InvalidTaskcontext提供task_result当前TaskResult与attempt尝试次数需要任务内部自报状态时用。执行与结果怎么查enqueue()把任务交给后端Immediate 后端当场执行完才返回真正的队列后端则先落队列、稍后执行此时拿到的是READY状态的结果。result send_verification_email.enqueue(userexample.com) # 指定时刻执行后端不支持延迟时会被拒绝 deferred generate_weekly_report.using(run_afterplanned_time).enqueue(2026, 34)状态、返回值与错误都挂在TaskResult上from django_tasks import TaskResultStatus if result.status TaskResultStatus.SUCCESSFUL: print(result.return_value) elif result.status TaskResultStatus.FAILED: print(result.errors[0].exception_class) print(result.errors[0].traceback)四个状态是READY入队待执行、RUNNING执行中、SUCCESSFUL、FAILED。注意return_value只在SUCCESSFUL时可用失败或未完成时访问会抛ValueError。跨请求或跨任务要拿结果先存result.id之后按 id 取回result_id result.id # 长度不超过 64 的字符串存库即可 # 任务侧取回id 与任务不匹配会抛 TaskResultMismatch later generate_weekly_report.get_result(result_id) # 后端侧取回任意任务的结果 from django_tasks import default_task_backend any_result default_task_backend.get_result(result_id)结果对象是快照后台状态变了就调result.refresh()刷新is_finished可快速判断是否已终态。监控卡在RUNNING的长任务时框架没有现成的批量查询接口稳妥做法是自己落一份 id 台账周期任务里逐个get_result再按started_at判断是否超时for rid in pending_ids: r default_task_backend.get_result(rid) if r.status TaskResultStatus.RUNNING and r.started_at now - one_hour: flag_stuck(r)后端与队列怎么搭配两个内置后端解决的是两个不同问题一个让功能先跑起来一个让测试不依赖执行。ImmediateBackend默认DummyBackend测试用行为当前线程立即执行只记录TaskResult不执行函数体执行位置请求线程内无执行supports_defer否是supports_async_task是是supports_get_result否结果不跨请求保留是但结果只在当前线程可见适用开发期、无队列时的兜底测试入队行为与参数用supports_*标志在运行时确认能力避免部署后才发现功能不可用from django_tasks import default_task_backend if default_task_backend.supports_defer: scheduled generate_weekly_report.using(run_afterplanned_time)四个标志的含义supports_defer能否run_after延迟执行、supports_async_task能否入队协程、supports_get_result能否从任意线程/进程按 id 取结果、supports_priority能否按优先级排序执行。⚠️ 两个内置后端都覆盖不了跨请求真异步Immediate 在请求线程里同步跑慢任务照样拖长响应。生产环境需要搭配真正基于队列的后端db 或 RQ 系扩展导入路径见django_tasks/backends/base.py的BaseTaskBackend选型时先查这四个标志。队列名是逻辑分组QUEUES写白名单后未列出的队列名在装饰或入队时抛InvalidTask防止打错字导致任务消失设为[]则关闭校验。典型分法TASKS { default: { BACKEND: django_tasks.backends.immediate.ImmediateBackend, QUEUES: [default, media, reports], }, }IO 密集任务压缩图片、发邮件放media计算任务周报放reports慢的挤不占快的。用信号监听任务生命周期三个信号对应任务的完整生命周期task_enqueued入队后、task_started执行前一刻、task_finished执行结束无论成败。handler 以关键字参数收到task_resultfrom django.dispatch import receiver from django_tasks import TaskResultStatus from django_tasks.signals import task_finished receiver(task_finished) def on_task_finished(sender, task_result, **kwargs): if task_result.status TaskResultStatus.FAILED: alert(task_result.task.module_path, task_result.errors[0].traceback)sender是后端类可以据此区分不同后端框架自身已内置了打日志的 handler失败任务会走logger.exception带出 traceback。埋点、通知、统计都挂在这里不用改业务代码。上生产前的避坑清单最后对照这十条过一遍能少掉进大多数坑任务函数定义在模块级嵌套函数抛InvalidTaskrun_after仅对supports_defer的后端有效否则任务立即执行结果保留时长取决于后端Immediate 的结果随进程结束消失长保留需换支持持久化的后端没有取消 API已入队任务撤不回靠任务内部检查取消标记实现业务级取消查失败任务看errors[0].exception_class与errors[0].tracebackreturn_value在FAILED/未完成时抛ValueError访问前先判statusQUEUES是白名单而非注册表写空列表[]才关闭校验priority只能是 -100 到 100 的整数后端不支持时设非零值直接抛错长任务自己落 id 台账 周期get_result做超时巡检attempts当前取值只有 0 或 1不要拿它当重试计数依赖官方任务框架的 API 边界很克制定义、入队、追踪、信号四件事闭环。把 django-tasks 当底座慢操作移出请求链路剩下的交给队列。【免费下载链接】django-tasksA backport of Djangos built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考