django-simpletask5

本项目由 opencode + deepseek-v4-flash 生成

一个轻量级的 Django 异步任务执行框架，提供声明式的任务模型、信号驱动的自动发布、Worker 进程异步执行，以及内置的 Cron 定时调度。

特性

• 声明式任务模型 — 继承 Task 模型即可定义任务，自动处理创建/更新/删除事件
• 信号驱动 — Django 信号自动拦截模型变更，发布 TaskExecution 到消息队列
• 自定义事件 — 通过 task.trigger('event_name') 触发任意事件
• 灵活的执行器映射 — 不同事件可绑定不同的执行器类
• 队列路由 — 不同事件可路由到不同优先级队列
• 重试与超时 — 失败自动重试（指数退避），支持 Pickup 超时检测；运行中孤儿任务由执行锁回收
• 拒绝回退 — 执行器可通过 RejectException 拒绝当前任务，消息自动重新入队等待下次调度
• 加密字段 — 敏感数据自动加密存储
• Cron 调度 — 内置 crontab 守护进程，支持代码注册与数据库覆盖
• 归档统计 — 已完成执行记录自动归档为加密 JSONL，并生成日统计
• Worker 注册中心 — 基于 Redis 的 Worker 心跳与状态追踪
• Django Admin 集成 — 完整的后台管理界面

依赖

• Python >= 3.10
• Django >= 4.2（兼容至 5.2.x）
• Kombu >= 5.6.2（消息队列，支持 RabbitMQ/Redis/内存）
• Redis >= 7.4.0（分布式锁、Worker 注册中心）
• 详见 pyproject.toml

安装

pip install django-simpletask5

django-simpletask5 使用 django-app-requires 自动管理依赖的 app（如 django_safe_fields），无需手动添加到 INSTALLED_APPS。只需在 settings.py 中调用一次 patch：

from django_app_requires import patch_all as django_app_requires_patch_all
django_app_requires_patch_all()

然后在 INSTALLED_APPS 中添加：

INSTALLED_APPS = [
    ...
    'django_simpletask5',
]

配置分布式锁（最小可用配置，完整配置项见下方「配置说明」）：

DJANGO_SIMPLETASK_LOCK_CONFIG = {
    'global_lock_engine_class': 'globallock.redis_global_lock.RedisGlobalLock',
    'global_lock_engine_options': {
        'host': '127.0.0.1',
        'port': 6379,
        'db': 0,
    },
}

运行迁移：

python manage.py migrate django_simpletask5

配置说明

以下给出面向项目集成的完整 settings.py 配置示例。除 django-simpletask5 相关配置外，其余 Django 默认配置项（DATABASES、MIDDLEWARE、TEMPLATES、ROOT_URLCONF 等）按项目自身需要配置，此处以 ... 省略。配置原则逐项见后文「配置项与配置原则」表。

import os
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent

SECRET_KEY = os.environ.get('DJANGO_SECRET_KEY', 'change-me-in-production')
DEBUG = os.environ.get('DJANGO_DEBUG', 'False') == 'True'
ALLOWED_HOSTS = ['*']

# ... 数据库（DATABASES）、中间件、模板、URL 等 Django 默认配置 ...

INSTALLED_APPS = [
    # django-simpletask5 的依赖 app（django_safe_fields 等）
    # 由 django-app-requires 自动补全，无需手动添加
    'django_simpletask5',
    # ... 你的业务 app ...
]

# ── 消息队列（MQ）────────────────────────────────────────────
# 生产使用 RabbitMQ 或 Redis；开发调试可用内存队列
DJANGO_SIMPLETASK_BROKER_URL = 'amqp://guest:guest@127.0.0.1:5672//'
DJANGO_SIMPLETASK_BROKER_HEARTBEAT = 60             # 发布连接池的 broker 心跳（秒）
DJANGO_SIMPLETASK_MAX_REQUEUE_DELIVERIES = 100      # AMQP 消息最大重投次数，超过后丢弃
# 若不设 BROKER_URL，则按以下参数组合：
# DJANGO_SIMPLETASK_MESSAGE_QUEUE_TYPE = 'rabbitmq'   # 'memory' | 'rabbitmq' | 'redis'
# DJANGO_SIMPLETASK_RABBITMQ_HOST = '127.0.0.1'
# DJANGO_SIMPLETASK_RABBITMQ_PORT = 5672
# DJANGO_SIMPLETASK_RABBITMQ_USER = 'guest'
# DJANGO_SIMPLETASK_RABBITMQ_PASSWORD = 'guest'
# DJANGO_SIMPLETASK_REDIS_MQ_HOST = '127.0.0.1'       # MESSAGE_QUEUE_TYPE='redis' 时
# DJANGO_SIMPLETASK_REDIS_MQ_PORT = 6379
# DJANGO_SIMPLETASK_REDIS_MQ_DB = 0
# DJANGO_SIMPLETASK_REDIS_MQ_PASSWORD = None          # Redis MQ 需认证时设置

# ── 分布式锁与 Worker 注册中心（Redis）────────────────────────
# 默认锁引擎 DjangoRedisGlobalLock 复用 CACHES['default']，需为 Redis 缓存
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'OPTIONS': {'CLIENT_CLASS': 'django_redis.client.DefaultClient'},
    },
}
# 如需独立连接参数（锁与 Worker 注册中心共用同一 host/port/db/password）：
# DJANGO_SIMPLETASK_LOCK_CONFIG = {
#     'global_lock_engine_class': 'globallock.redis_global_lock.RedisGlobalLock',
#     'global_lock_engine_options': {'host': '127.0.0.1', 'port': 6379, 'db': 0,
#                                    'password': '<redis 密码，无需认证可省略>'},
# }
# 注：globallock>=0.2.0 已取消按 timeout 自动释放的概念。锁的有效期由
# lockman/lock 作用域自行管理，作用域未结束前锁持续有效（内置看门狗自动续期），
# 锁在 `with` 块退出或进程消亡时自动释放。已移除 DJANGO_SIMPLETASK_LOCK_TIMEOUT
# 与 DJANGO_SIMPLETASK_LOCK_RENEW_INTERVAL，请勿再配置这两项。

# ── Key 前缀与队列路由 ───────────────────────────────────────
DJANGO_SIMPLETASK_KEY_PREFIX = 'myproject'           # Redis key 前缀，默认取 ROOT_URLCONF 首段
DJANGO_SIMPLETASK_DEFAULT_QUEUE = 'myproject:django-simpletask5:queue:default'
DJANGO_SIMPLETASK_HIGH_PRIORITY_QUEUE = 'myproject:django-simpletask5:queue:high_priority'

# ── 执行与重试 ───────────────────────────────────────────────
DJANGO_SIMPLETASK_DEFAULT_MAX_RETRIES = 3
DJANGO_SIMPLETASK_DEFAULT_TIMEOUT_SECONDS = 600          # 执行超时字段（秒，当前不主动中断）
DJANGO_SIMPLETASK_DEFAULT_PICKUP_TIMEOUT_SECONDS = 3600  # Pickup 超时（秒）

# ── 发布重试 ────────────────────────────────────────────────
DJANGO_SIMPLETASK_PUBLISH_RETRY_POLICY = {
    'max_retries': 2,
    'interval_start': 0.2,
    'interval_step': 0.3,
    'interval_max': 1.0,
}
DJANGO_SIMPLETASK_PUBLISH_RETRY_BASE_DELAY = 1.0      # 应用层第 1 次重试等待（秒）
DJANGO_SIMPLETASK_PUBLISH_RETRY_MAX_DELAY = 5.0       # 应用层重试等待上限（秒）

# ── Worker / 日志运维参数 ────────────────────────────────────
DJANGO_SIMPLETASK_WORKER_HEARTBEAT_INTERVAL = 60      # Worker 心跳/巡检间隔（秒）
DJANGO_SIMPLETASK_SHUTDOWN_TIMEOUT = 30               # 每个 worker 线程的优雅停机等待（秒）
DJANGO_SIMPLETASK_LOG_INTERVAL = 300                  # 存活/注册失败等日志节流间隔（秒）
DJANGO_SIMPLETASK_PUBLISH_LOG_INTERVAL = 60           # 发布/回收失败日志节流间隔（秒）

# ── 归档与统计 ──────────────────────────────────────────────
# 归档文件通过 TaskExecutionArchive.file 写入 MEDIA_ROOT/django_simpletask5_archives/%Y/%m/
MEDIA_ROOT = os.path.join(BASE_DIR, 'media')
DJANGO_SIMPLETASK_ARCHIVE_SALT = 'change-me-archive-salt'  # 归档加密盐值
DJANGO_SIMPLETASK_ARCHIVE_RETENTION_DAYS = 7

# ── Cron ────────────────────────────────────────────────────
DJANGO_SIMPLETASK_CRONJOB_AUTO_SYNC = True            # 自动同步代码注册的 Cron

# ── 安全：脚本执行器与字段加密 ───────────────────────────────
DJANGO_SIMPLETASK_ENABLE_SCRIPT_EXECUTORS = False     # 脚本执行器开关（生产保持关闭）
DJANGO_SIMPLETASK_SCRIPT_WHITELIST = []               # 脚本白名单
DJANGO_SIMPLETASK_FIELD_CIPHER_CLASS = None           # 自定义加密器类
DJANGO_SIMPLETASK_RESULT_PASSWORD = os.environ.get('DJANGO_SIMPLETASK_RESULT_PASSWORD')
DJANGO_SIMPLETASK_ERROR_MSG_PASSWORD = os.environ.get('DJANGO_SIMPLETASK_ERROR_MSG_PASSWORD')
DJANGO_SIMPLETASK_CONTEXT_PASSWORD = os.environ.get('DJANGO_SIMPLETASK_CONTEXT_PASSWORD')

# ── 可选：Admin Dashboard ────────────────────────────────────
# DJANGO_ADMIN_DASHBOARDS = {
#     'admin:app_list': {'django_simpletask5': 'django_simpletask5.dashboards.TaskDashboard'},
# }

# 必须最后调用：自动补全 django-simpletask5 依赖的 app
from django_app_requires import patch_all as django_app_requires_patch_all
django_app_requires_patch_all()

配置项与配置原则

注意：DJANGO_SIMPLETASK_ENABLE_SCRIPT_EXECUTORS 设为 True 后，还需为用户/组授予 django_simpletask5 | Cron job | Can use script executors 权限，非超级管理员无法创建 Python/Shell 脚本执行器类型的定时任务。

发布可靠性说明：应用层 _publish_message 最多尝试 3 次（首次 + 2 次重试），失败等待按 PUBLISH_RETRY_BASE_DELAY 指数增长并受 PUBLISH_RETRY_MAX_DELAY 限制；Kombu Producer 内部还会按 PUBLISH_RETRY_POLICY 重试。最终仍失败时，safe_publish_message() 会把非终态 execution 标记为 failed，不会让调用方因 broker 异常崩溃。publish_message() 保留异常上抛语义，供需要自行处理错误的调用方使用。

快速开始（从零到一集成）

本框架的集成分三步：定义任务模型 → 编写执行器 → 启动 Worker。完整可运行的参考实现见仓库中的 django_simpletask5_example 应用，包含 OrderTask 模型、创建/更新/删除/退款四个执行器及内置 Cron 任务。

1. 定义任务模型

继承 django_simpletask5.models.Task 即可获得信号驱动的异步能力。模型的保存（新增/更新）与删除会被自动拦截并发布消息，无需手写信号：

from django.db import models
from django_simpletask5.models import Task

class OrderTask(Task):
    order_id = models.CharField(max_length=64, unique=True)
    customer_name = models.CharField(max_length=128)
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    status = models.CharField(max_length=32, default='pending')

    # 事件 → 执行器映射（详见「事件处理机制」）
    executor_class = {
        'create': 'myapp.executors.OrderCreateExecutor',
        'update': 'myapp.executors.OrderUpdateExecutor',
        'delete': 'myapp.executors.OrderDeleteExecutor',
        'refund': 'myapp.executors.OrderRefundExecutor',   # 自定义事件，由 request_refund 通过 self.trigger('refund') 手动触发
    }

    # 事件 → 队列路由（未映射的事件自动落到 DJANGO_SIMPLETASK_DEFAULT_QUEUE）
    simpletask_queue = {
        'create': 'myproject:django-simpletask5:queue:high_priority',
        'update': 'myproject:django-simpletask5:queue:default',
    }

    # 仅当这些字段发生变化时才触发 update 事件（详见「事件处理机制」）
    trigger_update_fields = ['customer_name', 'amount', 'status']

    def request_refund(self, amount):
        return self.trigger('refund', extra_context={'refund_amount': str(amount)})

要点：

• 主键为 Django 默认自增 id。Task 不再内置 task_id 字段，业务标识由业务模型自行定义（自定义主键或自然键均可）
• 保存 / 删除 / 批量创建自动触发事件，详见「事件处理机制」
• 事件上下文、update 字段判定、delete 拦截等行为的完整说明见对应章节

2. 编写执行器

执行器继承 BaseExecutor，实现 execute() 方法即可：

# myapp/executors.py
from django_simpletask5.executors.base import BaseExecutor
from django_simpletask5.models import TaskExecution

class OrderCreateExecutor(BaseExecutor):
    timeout_seconds = 300               # 记录预期执行上限（当前不主动中断，可选）
    pickup_timeout_seconds = 1800       # Pickup 超时 30 分钟（可选）

    def execute(self, execution: TaskExecution, task=None) -> str | None:
        context = execution.get_context_dict()   # 事件上下文
        order = context['data']                  # create 事件的模型字段快照
        # ... 你的业务逻辑（调用外部系统、发消息等）...
        return 'ok'                              # 返回值存入 execution.result

执行器接口、批量任务、异常重试、拒绝回退等完整说明见「应用开发：执行器」。

3. 启动 Worker

python manage.py django_simpletask_executor --workers 4

• 默认监听 default 队列；high_priority 队列需单独启动 Worker 处理
• 可用 --service myapp.executors.OrderCreateExecutor 只处理指定执行器
• 可用 --debug 输出完整异常堆栈与 DEBUG 日志
• Worker 线程由 WorkerSupervisor 守护：异常退出自动重启、保持目标 Worker 数量；Redis/MQ 断连自动重连

4. 启动 Cron 调度（可选）

python manage.py django_simpletask_crontab

5. 触发自定义事件

在模型方法或业务代码中调用 trigger() 即可发布任意事件：

order = OrderTask.objects.get(order_id='ORD-001')
order.trigger('refund', extra_context={'refund_amount': '50.00'})

6. 验证

启动 Worker 后新增一条 OrderTask，观察数据流转：

OrderTask.objects.create(order_id='ORD-001', customer_name='Test', amount=Decimal('100.00'))

• 数据库 TaskExecution 表新增一条 create 事件、pending 状态的记录
• Worker 消费后状态变为 running，执行完成变为 success，result 保存执行器返回值
• Django Admin 的 Dashboard 可查看执行统计与 Worker 状态

事件处理机制

内置事件：create / update / delete

框架通过 Django 信号自动监听所有 Task 子类的保存与删除，无需手动发布：

执行器通过 execution.get_context_dict() 获取以上上下文。其中 task_id 是业务对象主键的字符串表示（如 "42"、UUID 字符串或自定义键）；业务对象反查见「模型与生命周期」。

update 事件的字段变更判定

update 事件与保存动作的关系：框架每次保存已存在的实例时，通过 pre_save/post_save 信号对比旧值快照与新值，计算发生变化的字段列表（changed_fields）。默认只要任一字段变化就发布 update 事件——这往往会产生大量非预期消息。例如某次保存只改了一个内部计数或审计字段，下游系统却会收到一次同步通知。

为此可用类属性控制「哪些字段变化才值得触发 update 事件」：

class OrderTask(Task):
    # 白名单：仅当这些字段之一发生变化时才发布 update 事件
    trigger_update_fields = ['customer_name', 'amount', 'status']

    # 或黑名单：这些字段变化不发布 update 事件，其余字段变化照常发布
    # trigger_update_blacklist = ['sync_count', 'updated_by']

白名单与黑名单二选一，同时配置时以白名单为准。无论何种方式，值未实际变化（如调用 save() 但所有字段值相同）的保存都不会触发 update 事件。changed_fields 中即包含本次实际发生变化的字段名列表，执行器可据此做增量处理。

delete 事件的拦截与确认删除

Task.delete() 被框架拦截，不会立即删除数据库记录，而是发布 delete 事件消息。执行器处理完外部系统清理后，调用 execution.confirm_delete() 完成真实删除：

class OrderDeleteExecutor(BaseExecutor):
    def execute(self, execution: TaskExecution, task=None) -> str | None:
        # ... 先清理外部系统关联数据 ...
        execution.confirm_delete()   # 真正删除对应 Task 记录
        return 'deleted'

这样实现删除操作与外部系统清理的解耦，且支持失败重试。

confirm_delete() 与 execution.get_task() 都会通过 TaskExecution.task_id（业务主键字符串）反查业务对象，反查逻辑由业务模型上的 get_task_instance() 提供（见「模型与生命周期」）。

事件 → 执行器映射

每个事件消息最终要由一个执行器处理。executor_class 声明「事件名 → 执行器类路径」的分派表，可以是字符串（所有事件共用同一个执行器）或字典（按事件分派）：

executor_class = {
    'create':  'myapp.executors.OrderCreateExecutor',
    'update':  'myapp.executors.OrderUpdateExecutor',
    'delete':  'myapp.executors.OrderDeleteExecutor',
    'refund':  'myapp.executors.OrderRefundExecutor',   # 自定义事件
    'default': 'myapp.executors.OrderDefaultExecutor',  # 兜底执行器（可选）
}

执行器解析顺序：命中事件名（如 refund 映射到 OrderRefundExecutor）→ default → create → 自动派生 app_label.tasks.{ClassName}{Event.capitalize()}Executor（以上都未命中时的兜底路径）。

注意：executor_class 只负责「事件 → 由谁执行」的映射，与事件由谁触发无关。内置事件（create/update/delete）由信号自动触发；自定义事件（如 refund）必须由业务代码调用 trigger('refund') 手动触发，详见下方「自定义事件」。

事件 → 队列路由

每个事件消息发布到哪个队列，由 simpletask_queue 决定。可以是字符串（所有事件同一队列）或字典（按事件路由）：

simpletask_queue = {
    'create': 'myproject:django-simpletask5:queue:high_priority',
    'update': 'myproject:django-simpletask5:queue:default',
}

队列解析顺序：命中事件名 → default → 全局默认队列 DJANGO_SIMPLETASK_DEFAULT_QUEUE。

以上面只配置 create/update 两个键为例，各事件的落队结果：

即：只映射了部分事件时，其余未映射的事件自动落到 DJANGO_SIMPLETASK_DEFAULT_QUEUE。如需统一兜底到某个队列，可显式添加 'default' 键。

自定义事件

与信号自动触发的内置事件不同，自定义事件（如 refund）不会自动触发，需要业务代码调用 Task.trigger(event, extra_context=None) 手动发布：

order = OrderTask.objects.get(order_id='ORD-001')
order.trigger('refund', extra_context={'refund_amount': '50.00'})
# 触发后流程：
# trigger('refund') → 创建 trigger_event='refund' 的 TaskExecution
# → 按 simpletask_queue 路由发布消息 → Worker 按 executor_class['refund']
#   加载 OrderRefundExecutor 执行

• 事件名任意：trigger() 的第一个参数即事件名，需与 executor_class / simpletask_queue 字典中的键保持一致，否则走 default 或兜底路径
• 执行器：按「事件 → 执行器映射」解析
• 队列：按「事件 → 队列路由」解析
• 上下文：extra_context 会合并进执行器可见的事件上下文，执行器侧读取 context['refund_amount']：

class OrderRefundExecutor(BaseExecutor):
    def execute(self, execution, task=None):
        context = execution.get_context_dict()
        refund_amount = context['refund_amount']   # == '50.00'
        ...

抑制异步消息

临时禁止某个实例发布消息（_skip_publish 为临时实例属性，不会持久化到数据库），四种使用方式：

order._skip_publish = True; order.save()         # 方式一：实例属性
order.skip_publish().save()                      # 方式二：链式方法
order.save(skip_publish=True)                    # 方式三：save 参数
OrderTask.objects.create(..., skip_publish=True)  # 方式四：构造参数（推荐）

不会创建 TaskExecution，也不会发送异步消息。

应用开发：执行器

BaseExecutor 接口

execute 签名与返回值

def execute(self, execution: TaskExecution, task=None) -> str | None:
    context = execution.get_context_dict()   # 事件上下文（见「事件处理机制」）
    # ... 业务逻辑 ...
    return 'result'

• 事件上下文：execution.get_context_dict() 返回反序列化后的 JSON 上下文
• 返回值：单个任务时直接返回（字符串或 JSON 字符串）；返回 None 时 result 保持为空

批量任务

重写 get_tasks() 与 get_tasks_count()，一个执行器即可处理多条子任务：

class BatchSyncExecutor(BaseExecutor):
    def get_tasks_count(self, execution):
        return len(self._ids(execution))

    def get_tasks(self, execution):
        for _id in self._ids(execution):
            yield _id

    def _ids(self, execution):
        return execution.get_context_dict().get('ids', [])

    def execute(self, execution, task=None):
        self._sync(task)   # task 为 get_tasks() 产出的单个子任务
        return 'done'

批量执行时 execution.result 聚合为 {"results": [...], "total": N}，每完成一个子任务更新 done_tasks_count 便于进度展示。

异常与状态流转

执行器抛出的异常由框架统一处理：

拒绝回退

执行器遇到临时条件不满足（如资源上限），抛出 RejectException 拒绝本次执行：

from django_simpletask5.executors.exceptions import RejectException

class MyExecutor(BaseExecutor):
    def execute(self, execution: TaskExecution, task=None) -> str | None:
        if not self._can_proceed():
            raise RejectException('Resource limit reached, try again later')

框架捕获后：

• 调用 message.reject(requeue=True) 将消息放回队列尾部，下次重新调度
• TaskExecution 状态恢复为 pending，started_at 清空，等待下次 Worker 消费
• 不消耗重试次数
• 刷新 expire_time（使用 Pickup 超时），防止被 StatusCheckExecutor 误回收

超时配置

class OrderCreateExecutor(BaseExecutor):
    timeout_seconds = 300            # 记录预期执行上限（当前不主动中断）
    pickup_timeout_seconds = 1800    # Pickup 超时 30 分钟

未配置时使用全局默认值 DJANGO_SIMPLETASK_DEFAULT_TIMEOUT_SECONDS（600 秒）与 DJANGO_SIMPLETASK_DEFAULT_PICKUP_TIMEOUT_SECONDS（3600 秒）。

注意：当前版本只有 pickup_timeout_seconds 会被框架自动判定并触发 timeout；timeout_seconds 不会中止正在运行的执行器。需要严格执行时限时，请在 execute() 内部自行实现超时或使用可取消的子进程。

模型与生命周期

Task 抽象模型

所有业务任务模型继承 Task：

• 主键为 Django 默认自增 id，无需自定义；不内置 task_id 字段，业务标识由业务模型按需定义
• 提供 trigger(event, extra_context)、skip_publish()、get_executor_class_path(event)、get_queue_name(event) 等方法
• 保存、删除、批量创建自动触发事件（见「事件处理机制」）

业务对象反查：get_task_instance()

TaskExecution.task_id 存的是业务对象主键的字符串表示（CharField）。需要反查业务对象时（execution.get_task() / execution.confirm_delete()），框架先通过 task_model 解析出模型类，再调用该类的 get_task_instance(task_id) 完成查询。

默认实现在 Task 基类中按主键 pk 查询，并通过内置的 _coerce_task_pk() 兼容常见主键类型：

• AutoField / BigAutoField / SmallAutoField — 字符串转为 int
• UUIDField — 字符串转为 uuid.UUID
• CharField / TextField 等字符串主键 — 原样使用

task = execution.get_task()          # 返回业务对象（反查不到返回 None）
execution.confirm_delete()          # 反查并真实删除业务对象

业务模型使用自定义主键或非主键自然键时，重载 get_task_instance() 即可，框架会自动改走你的实现：

class OrderTask(Task):
    code = models.CharField(max_length=64, unique=True)   # 非主键自然键

    @classmethod
    def get_task_instance(cls, task_id):
        if task_id is None:
            return None
        return cls.objects.filter(code=task_id).first()

TaskExecution 状态机

每条事件消息对应一条 TaskExecution 记录，状态流转如下：

pending ──(Worker 消费)──▶ running ──(执行成功)──▶ success
   │                           │
   │                           ├─(异常且可重试)────────▶ retry ──(重新入队并被消费)──▶ running
   │                           ├─(异常且重试耗尽)──────▶ failed
   │                           └─(执行器异常退出/锁丢失)─▶ retry 或 failed（由 reclaim 回收）
   ├─(pickup 超时，从未被领取)──▶ timeout ──(RetryTimeout)──▶ retry
   ├─(用户取消)────────────────▶ canceled（终态）
   └─(发布最终失败)────────────▶ failed

取消任务

已发布但未执行（pending）的任务可通过 Admin 列表页勾选后执行 「Cancel selected pending tasks」 操作批量取消，状态置为 canceled。代码方式：

from django_simpletask5.models import TaskExecution

TaskExecution.objects.filter(
    execution_id='<execution_id>', status='pending',
).update(status='canceled')

canceled 为终态：即使消息仍留在 MQ 队列中，Worker 消费时也会直接跳过（不会执行），且不会被归档统计混入 failed。已进入 running 的任务无法取消。

重试与超时

• 重试：max_retries（默认 3）控制最大重试次数，retry_count 累计已重试次数；重试消息重新发布到原队列
• Pickup 超时（pickup_timeout_seconds）：唯一由框架自动判定的超时。pending 超过 expire_time 后由 StatusCheckExecutor 标记为 timeout，也用于拒绝回退后的重新调度保护；Pickup 超时不消耗重试次数
• 执行超时字段（timeout_seconds）：写入 TaskExecution.timeout_seconds，并在任务进入 running 时用于计算 expire_time。当前版本不会因该时间到达而主动中断 running 任务；如果执行器 hang 住，需在执行器内部实现超时/取消机制
• 超时检测与恢复：StatusCheckExecutor 每分钟把超过 pickup 期限的 pending 记录标记为 timeout；RetryTimeoutExecutor 每分钟把 timeout 记录重新入队，实际运行过的记录消耗一次重试次数。running 孤儿任务由 crontab 基于执行锁过期回收，而不是按 expire_time 判超时

其他模型

• CronJob — 定时任务注册表，支持代码注册与数据库覆盖（见「Cron 任务」）
• TaskExecutionArchive — 归档记录，终态执行记录归档为加密 JSONL（见配置「归档」）
• TaskExecutionStat — 按日期、executor_class、trigger_event 聚合的日统计

Cron 任务

在代码中定义 Cron 任务

在任意 app 的 cronjobs.py 文件中使用 register_cronjob 注册，并在该 app 的 AppConfig.ready() 中导入这个模块；框架不会自动发现各 app 下的 cronjobs.py：

# myapp/cronjobs.py
from django_simpletask5.core.cronjob_registry import register_cronjob

register_cronjob(
    uid='health_check',
    display_name='Health Check',
    cron_expression='*/5 * * * *',
    executor_class='django_simpletask5.executors.simple_request.SimpleRequestExecutor',
    context={
        'url': 'https://example.com/health',
        'method': 'GET',
        'timeout': 10,
    },
    description='定期健康检查',
)

# myapp/apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        import myapp.cronjobs  # noqa: F401

从代码同步到数据库

注册的 Cron 任务需要同步到数据库才会生效。django_simpletask_crontab 运行期间每个 tick 都会自动同步（默认开启，可通过 DJANGO_SIMPLETASK_CRONJOB_AUTO_SYNC = False 关闭自动同步），也可以手动执行：

python manage.py django_simpletask_sync_cronjobs

手动同步命令使用 sync_cronjobs(force=True)，即使关闭了自动同步也会执行。同步后，用户可以在 Django Admin 中查看和修改 Cron 任务，被手动修改过的任务不会在后续同步中被覆盖（is_modified_by_user 标记保护）。单个 CronJob 创建失败只记录日志，不会中断其它任务同步。

启用/停用（is_active）的所有权独立跟踪，规则与 is_modified_by_user 对称——手动改过就一直归你，直到显式重置：

• 用户未手动启停过的任务，is_active 会按代码注册值同步（例如 simpletask_ping_pong 注册为 is_active=False，默认保持停用）。
• 用户手动暂停/启用后，会置 is_active_modified_by_user 标记，自动同步不再覆盖该状态，但不影响其它定义字段（cron_expression 等仍照常同步）。
• 该标记不会自动清除；需要让它重新跟随代码时，使用管理后台的“从代码定义重置”（reset_from_code），或在编辑页取消勾选“保留手动启用状态”。
• 这样手动启停的状态可稳定保持，不会因代码默认值变化而悄悄被改写。

注意：cron 表达式按 Django 的 TIME_ZONE 设置对应的本地时间解释。USE_TZ=True 时仍以本地时间计算触发时刻（存储为 UTC），例如 TIME_ZONE = 'Asia/Shanghai' 下 30 14 * * * 表示每天 14:30 本地时间触发。

内置执行器

开发与测试

项目提供 start-test-services.sh 一键启动测试所需的 Redis 与 RabbitMQ 服务。脚本优先检测本地已拉取的镜像（避免依赖 docker.io 上游镜像），也可通过 DJANGO_SIMPLETASK_REDIS_IMAGE / DJANGO_SIMPLETASK_RABBITMQ_IMAGE 环境变量指定，默认使用 6379/5672 端口（被占用时自动选取空闲端口）：

./start-test-services.sh start    # 启动服务并等待就绪
./start-test-services.sh status   # 查看容器与端口状态
./start-test-services.sh stop     # 停止并清理容器

运行测试：

python3 -m pytest django_simpletask5_example/tests

测试套件覆盖 Redis、DB、MQ 连接中断后的自动恢复场景，详见 django_simpletask5_example/tests/test_resilience_recovery.py。

测试输出约定：单元测试中对预期会触发的错误（如无效 UUID 校验、毒消息丢弃、消费者断连、执行器加载失败等）所输出的错误日志，必须在测试内屏蔽，避免污染测试输出。统一使用 django_simpletask5_example.tests.suppress_expected_errors 上下文管理器（整体禁用日志）或 patch.object(..., 'logger')（断言同时校验日志）进行屏蔽。

架构

Task 模型变更 → Django 信号 → 创建 TaskExecution 并发布到消息队列
                                                    ↓
Worker 消费消息 → 获取分布式锁 → 加载执行器 → 执行并保存结果
                                                    ↓
失败时自动重试/拒绝时重新入队，完成后归档

Releases

0.2.9

• 启用状态所有权独立跟踪 — 新增 CronJob.is_active_modified_by_user 字段与迁移 0014，将“启用/暂停状态”与“定义字段”的所有权彻底分离。用户手动启停（快捷按钮 / 批量操作 / 编辑表单）后置该标记，自动同步（sync_cronjobs）不再覆盖 is_active；未手动改过的任务仍按代码注册值同步。暂停只锁 is_active，cron_expression 等定义字段照常同步
• 显式释放所有权 — 该标记不会自动清除；需要让任务重新跟随代码时，使用管理后台“从代码定义重置”（reset_from_code），或在编辑页取消勾选“保留手动启用状态”
• 保留列表分页/过滤条件 — 修复暂停/运行按钮提交后重定向丢失查询串的问题：按钮表单 action 及视图重定向均携带当前 changelist 的 GET 参数，操作后刷新仍停留在原分页、过滤器与搜索条件
• 消息发布可靠性增强 — Kombu 发布复用连接池并支持 broker heartbeat；publish() 返回 queue / message_count / consumer_count 队列统计；新增可配置发布重试策略和递增重试延迟；发布异常时自动丢弃损坏连接；新增公开的 publish_message() 与兜底的 safe_publish_message()，最终发布失败会将 execution 标记为 failed
• 发布日志节流 — 发布、重试投递、孤儿任务回收和 CronJob 触发失败统一接入日志节流，避免 broker/Redis 长时间不可用时刷屏；重投日志同时记录队列积压与消费者数量
• Worker 与回收健壮性 — Worker 注册表 Redis 客户端改为公开 get_redis_client()；修正 Worker 清理 key 的 glob 为 worker:{pid}:*；reclaim._lock_config() 与 get_lock_manager() 对齐，只有 None 才回退默认锁配置；回收扫描前后关闭旧数据库连接，减少长时间运行后的连接失效
• 日志初始化修复 — setup_console_logging() 不再把 FileHandler 误判为控制台 handler，避免已有文件日志时跳过控制台输出
• CronJob 同步健壮性 — sync_cronjobs(force=True) 支持手动同步命令绕过 DJANGO_SIMPLETASK_CRONJOB_AUTO_SYNC；单条 CronJob 创建失败不再中断其它任务同步；django_simpletask_sync_cronjobs 命令改为强制同步
• API 命名整理 — _CRONJOB_REGISTRY → CRONJOB_REGISTRY、_compute_next_run_time() → compute_next_run_time()、_get_broker_url() → get_broker_url()、_model_to_dict() → model_to_dict()、_DeleteIntercepted → DeleteIntercepted、_get_redis_client() → get_redis_client()；依赖旧私有名称的扩展或测试需同步调整
• 可观测性入口公开 — Dashboard 使用的 Worker 统计入口由 _get_worker_stats 改为公开的 get_worker_stats()；消费者线程心跳改为读取 DJANGO_SIMPLETASK_WORKER_HEARTBEAT_INTERVAL，Worker 注册、Redis 降级和日志节流逻辑补充了完整说明
• 注释与文档校准 — 为整个 django_simpletask5 生产包补充中文模块/类/函数/分支注释，并修正归档路径、归档清理、running/timeout 状态流转、Admin 只读性、脚本白名单、连接池 heartbeat、优雅退出等待时间等与实际行为不一致的说明
• 迁移与翻译 — 新增迁移 django_simpletask5.0014_cronjob_is_active_modified_by_user，同步更新简体中文翻译与迁移中的字段 help_text
• 测试 — 新增 test_logging_utils.py；扩充消息发布、消息队列、CronJob 同步、Admin 启停、Reclaim、Worker 恢复、Redis 锁与集成测试，覆盖最终发布失败、队列积压统计、force 同步、启停所有权、查询串保留、空锁配置和 Worker 清理 glob 等场景
• 环境要求 — 代码使用 X | None 联合类型，最低 Python 版本明确为 3.10；同时将 Django 最低版本明确为 4.2，并补充 Django>=4.2 依赖声明
• 版本 — pyproject.toml 版本更新为 0.2.9

0.2.8

• 移除 Task 冗余的 task_id 字段 — TaskExecution.task_id 由 UUID 改为 CharField(max_length=255)，直接用于存放业务对象主键字符串（str(instance.pk)），不再维护一份与业务主键重复的 UUID。消息上下文中的 task_id 字段名不变，值语义从 UUID 变为业务主键字符串（如 "42"）
• 业务对象反查下沉到模型层 — 新增 Task.get_task_instance(task_id) 类方法作为默认反查实现（按 pk 查询，内置 _coerce_task_pk() 兼容 int / UUID / 字符串主键）；TaskExecution.get_task() / confirm_delete() 改为解析 task_model 后委托给该方法。业务模型使用自定义主键或自然键时重载 get_task_instance() 即可
• 新增迁移 — django_simpletask5.0013_alter_taskexecution_task_id（字段改 CharField）、django_simpletask5_example.0003_remove_ordertask_task_id
• 破坏性变更: 依赖 TaskExecution.task_id 为 UUID 类型的外部消费方需适配为业务主键字符串；业务 Task 子类若曾使用内置 task_id 字段需移除

0.2.7

• 运行中任务回收机制重构 — 取消原先按 expire_time 将 running 直接判为超时的语义；StatusCheckExecutor 仅兜底回收长期未被 Worker 获取的 pending。新增 core/reclaim，由 crontab 每 ~5 秒基于执行锁探测运行中任务：锁已过期（执行器异常退出）即快速重投或直接置失败，降低孤儿任务滞留时间
• 重试不再搁浅任务 — 失败重试先落库 retry 再发布重试消息，发布失败则 requeue 当前消息，杜绝任务永久停在无人处理的 retry 状态；KombuQueue 依据 AMQP x-death 上限丢弃毒消息，避免无限 requeue 造成的队头阻塞与空转
• 消费者重连提示 — Worker 断连重连成功后输出 Consumer connection restored 恢复日志，便于运维确认消费通道已恢复
• 安全与健壮性加固 — broker 账号/密码按百分号编码、脚本白名单按路径段边界匹配、signals 快照按实例隔离，覆盖相关回归测试
• 新增测试约定 — 单元测试中对预期触发（如无效 UUID 校验、毒消息丢弃、消费者断连、执行器加载失败等）产生的错误日志必须屏蔽输出，统一使用 django_simpletask5_example.tests.suppress_expected_errors 或替换 logger，保证测试输出干净可读；本次同步清理全部此类预期错误日志

0.2.6

• 新增 canceled 状态 — 取消已发布但未执行的任务改为独立的终态 canceled（原复用 failed），Worker 消费到已取消消息时直接跳过；Admin 取消操作、归档/清理/统计（新增 canceled_count）、Dashboard 状态图均纳入该状态，取消与执行失败可清晰区分
• WorkerSupervisor 守护线程 — 统一的 Worker 线程保活机制，周期性心跳并自动重启异常退出的 Worker 线程，保持目标 Worker 数量，提升长时间运行稳定性
• 韧性/恢复测试套件 — 新增 test_resilience_recovery.py，模拟 Redis、DB、MQ 连接中断后的自动恢复：Worker 启动时 Redis 注册失败重试、消费者断连自动重连、锁引擎中断后消息重入队并在 Redis 恢复后继续处理、DB 瞬断重入队、执行器异常标记 failed/retry、crontab 调度器 Redis 恢复检测等
• 测试套件稳定性修复 — test_worker_registry 隔离全局 _active_workers 消除跨用例污染；test_integration 动态探测本地可用的 Redis/RabbitMQ 镜像（支持 DJANGO_SIMPLETASK_REDIS_IMAGE / DJANGO_SIMPLETASK_RABBITMQ_IMAGE 环境变量覆盖），并改为真实 AMQP 握手探测 RabbitMQ 就绪，避免新建容器连接被重置的偶发失败
• 测试服务脚本 — start-test-services.sh 默认使用 6379/5672 端口并动态检测本地镜像，提供 start / stop / status 子命令，便于一键启动测试依赖

0.2.5

• 修复 0010 迁移双主键冲突 — 修正迁移 0010_taskexecution_id_alter_taskexecution_execution_id 的操作顺序：先移除 execution_id 的主键约束（改为 unique=True），再添加自增 id 主键，避免在已有数据库上升级时因同时存在两个主键而失败。已有 0.2.4 数据可正常升级，id 自动回填，execution_id 保持唯一

0.2.4

• 破坏性变更: TaskExecution 主键由 execution_id（UUID）改为默认自增 id，execution_id 变为 unique=True 的唯一标识字段。已有数据库需要执行迁移 0010_taskexecution_id_alter_taskexecution_execution_id
• 定期存活日志 — Worker 与 Crontab 主循环在长时间无消息/无任务时按 DJANGO_SIMPLETASK_LOG_INTERVAL（默认 300 秒）定期输出存活日志，并统一带时间戳格式，便于运维确认进程未死循环
• MQ 日志降噪 — MQ 消费/发布异常及 Redis 连接失败改为 error/warning 级别且无堆栈（默认按间隔节流），避免刷屏；提供 --debug 时输出完整堆栈。注：此处所指的"锁续期失败"是 0.2.4 版本引入的自定义锁续期机制，自 globallock>=0.2.0 起该机制已移除，改由 globallock 内置看门狗在 lockman/lock 作用域内自动续期，无需（也无从）单独记录续期失败日志
• 新增配置 — DJANGO_SIMPLETASK_LOG_INTERVAL 控制死循环存活日志与异常节流日志的输出间隔（秒）

0.2.3

• Cron 时区修复 — cron 表达式改为按 Django TIME_ZONE 配置的本地时间解释，不再固定按 UTC 计算。USE_TZ=True 时使用 timezone.localtime(now()) 作为 croniter 基准时间，非 UTC 时区下定时任务不再偏移；USE_TZ=False 保持原有本地时间行为不变

0.2.2

• 抑制异步消息 — 新增 _skip_publish 机制，创建/更新/删除 Task 时可通过 skip_publish() 方法、save(skip_publish=True)、构造时传入 skip_publish=True 等方式禁止触发 RabbitMQ 消息通知
• 批量创建触发异步消息 — TaskQuerySet + TaskManager 接管 bulk_create，批量创建 Task 后自动生成 TaskExecution 并推送 MQ；子类覆盖 objects 时发出警告提示
• 分布式锁默认引擎 — 默认使用 DjangoRedisGlobalLock，复用 Django CACHES['default'] 配置（需为 Redis 缓存），无需手动配置 DJANGO_SIMPLETASK_LOCK_CONFIG
• Redis key 命名统一 — 所有 Redis key 格式规范为 {项目前缀}:django-simpletask5:{用途}:{id}，项目前缀从 ROOT_URLCONF 自动提取，也可通过 DJANGO_SIMPLETASK_KEY_PREFIX 显式覆盖
• CronJob 管理增强 — 新增 app_label 字段（自动检测），列表页展示并支持过滤
• 内置 CronJob 重命名 — Status Check → Task Timeout Detection，Retry Timeout → Retry Failed Tasks，Ping/Pong → Health Check 等
• Dashboard 重做 — 新增今日失败卡片，成功率按 90%/80% 分绿色/橙色/红色三级，卡片颜色统一使用 info 蓝色
• Pending/Running 显示修复 — 改为展示 pending / running 双值而非仅 pending
• 成功率显示优化 — (22/25) 与 88% 同行显示，字号缩小颜色淡化，高度对齐
• Redis 不可达降级 — Worker 数量显示 - 而非 0，区分无数据与异常
• 按钮优化 — 改用 Font Awesome 图标，统一 min-width 对齐，title 改为中文描述
• 翻译补充 — 补充 Run/Activate/Deactivate 及所有内置 CronJob 名称的中文翻译
• 新增依赖 — django-static-fontawesome 用于按钮图标
• 示例应用 — 示例 cronjobs 模块在 AppConfig.ready() 中导入确保注册生效

0.2.1

• 拒绝回退机制 — 新增 RejectException，执行器可抛出该异常拒绝当前任务，消息自动重新入队等待下次调度，不消耗重试次数
• 区分两种超时 — 引入 pickup_timeout_seconds（等待 Worker 消费超时）与 timeout_seconds（执行超时）两种超时配置，expire_time 改用 Pickup 超时计算，超时检测更精准
• 文档更新 — README 补充拒绝回退机制的使用说明和两种超时的详细解释

0.2.0

• 破坏性变更: 移除 Task 模型上的自定义主键，task_id 不再是主键字段，改为 unique=True 的唯一标识字段。所有 Task 子模型将自动获得 Django 默认的自增 id 主键。已有数据库需要迁移处理。
• 统计维度调整: TaskExecutionStat 的统计维度由 task_model 改为 executor_class，统一覆盖有模型任务和 Cron 定时任务两种场景。对应 API _update_stats() 按 executor_class 分组聚合。
• Admin 优化: CronJob 列表页移除 executor_class 列避免表格撑开；TaskExecutionStat 列表页新增 executor_class 列、created_at 设为只读修复详情页错误。
• 测试修复: 修复 TransactionTestCase + on_commit 在 SQLite 共享内存连接下因 TestCase 残留原子块导致回调不执行的兼容性问题；Redis MQ / RabbitMQ E2E 集成测试全部通过。

0.1.5

• 安全加固 — 新增 DJANGO_SIMPLETASK_ENABLE_SCRIPT_EXECUTORS 配置项（默认关闭），启用后才可执行 Python/Shell 脚本；新增 can_use_script_executors 权限，仅有此权限或超级管理员的用户才能在 Admin 中创建脚本执行器类型的 CronJob
• Bug 修复 — 修复 RetryTimeoutExecutor 中 lambda 晚绑定导致所有回调引用最后一个记录的问题
• Admin 安全 — 操作按钮（立即执行、启用/禁用）从 GET 参数改为 POST 请求，附带 CSRF 保护
• 性能优化 — RetryTimeoutExecutor 改为批量 update + 按 ID 分发消息推送
• Admin 增强 — TaskExecutionAdmin 新增 executor_class 过滤、error_message 搜索、批量重试失败任务、批量取消待处理任务
• 数据索引 — 添加 (status, expire_time) 和 (is_active, next_run_time) 复合索引；trigger_event、executor_class、task_id、created_at 添加单字段索引
• 代码规范 — bash_script.py、python_script.py、simple_request.py 补充缺失的 logger；移除废弃的 allow_tags 属性

0.1.4

• 修复打包 — pyproject.toml 添加 package-data 配置，打包时包含 locale po/mo 文件，修复 i18n 翻译不生效的问题
• 补充依赖 — requirements.txt 和 pyproject.toml 添加缺失的 django-checkbox-normalize、django-tabbed-changeform-admin 依赖
• 完善文档 — README 新增完整配置说明章节，涵盖消息队列（MQ）、分布式锁、队列路由、归档、Cron、安全等全部配置项

0.1.3

• Admin 界面大升级 — 集成 Tabbed ChangeForm 分页签展示字段，CronJob 后台支持直接执行/启用/禁用操作；TaskExecution 列表优化，execution_id 显示前8字符并支持点击复制，重试次数合并显示，全局字段中文翻译
• 归档管理增强 — 新增 django_simpletask_archive 管理命令，支持手动触发归档与过期清理；PurgeExpiredExecutor 定期自动清理过期记录；保留天数可配置；采用流式 AES-GCM 加密，大文件归档更高效
• Dashboard 修复 — 卡片链接使用 reverse 正确跳转，无执行数据时成功率显示 -
• 超时检测优化 — 模型新增 expire_time 字段，超时判定更准确高效

0.1.2

• 修复: 修复 Transaction-Message 排序竞态条件，使用 transaction.on_commit 确保消息在事务提交后才发送到队列

0.1.1

• 修复: 修复 DjangoSimpletask5Config 缺少 ready() 方法导致信号监听未注册的问题

0.1.0

这是 django-simpletask5 的首个正式版本。核心功能包括：

• 声明式任务模型 — 继承 Task 模型即可定义任务，自动处理创建/更新/删除事件
• 信号驱动自动发布 — Django 信号自动拦截模型变更，发布 TaskExecution 到消息队列
• Worker 异步执行 — 多 Worker 进程消费消息队列，支持队列路由和优先级
• Cron 定时调度 — 内置 crontab 守护进程，支持代码注册与数据库覆盖
• 重试与超时 — 失败自动重试（指数退避），支持超时检测
• 加密字段 — 敏感数据自动加密存储
• 归档统计 — 已完成执行记录自动归档为加密 JSONL，并生成日统计
• Worker 注册中心 — 基于 Redis 的 Worker 心跳与状态追踪
• Django Admin 集成 — 完整的后台管理界面与仪表盘

许可证

MIT