django-simpletask5 0.2.1

Lightweight async task execution framework for Django

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.8
• Django >= 3.2（兼容至 5.2.x）
• Kombu >= 5.3.0（消息队列，支持 RabbitMQ/Redis/内存）
• Redis >= 4.0.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

配置说明

所有配置项均为可选，设有合理的默认值。以下按功能分类说明。

消息队列（MQ）

默认使用内存队列（memory://），适合开发调试。生产环境建议切换为 RabbitMQ 或 Redis。

# 方式一：直接指定完整 Broker URL（优先级最高）
DJANGO_SIMPLETASK_BROKER_URL = 'amqp://guest:guest@127.0.0.1:5672//'

# 方式二：分别指定类型和连接参数
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'

# Redis 作为 MQ 时：
# DJANGO_SIMPLETASK_MESSAGE_QUEUE_TYPE = 'redis'
# DJANGO_SIMPLETASK_REDIS_MQ_HOST = '127.0.0.1'
# DJANGO_SIMPLETASK_REDIS_MQ_PORT = 6379
# DJANGO_SIMPLETASK_REDIS_MQ_DB = 0

分布式锁

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,
    },
}
DJANGO_SIMPLETASK_LOCK_TIMEOUT = 600  # 锁超时时间（秒）

队列路由

DJANGO_SIMPLETASK_DEFAULT_QUEUE = 'django_simpletask5.queue.default'          # 默认队列
DJANGO_SIMPLETASK_HIGH_PRIORITY_QUEUE = '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 超时（秒），等待 Worker 消费的时间限制

框架区分两种超时：

• 执行超时（timeout_seconds）— 任务进入 running 状态后，必须在指定时间内完成，否则被 StatusCheckExecutor 标记超时。适用于保护实际执行不 hang 死。
• Pickup 超时（pickup_timeout_seconds）— 任务创建后等待 Worker 消费的时间上限，也用于被拒绝回退后等待重新调度的保护。通常设得比执行超时更长，因为排队等待时间不可控。

归档

DJANGO_SIMPLETASK_ARCHIVE_PATH = 'django_simpletask5_archives'  # 归档文件存储子目录
DJANGO_SIMPLETASK_ARCHIVE_SALT = 'django-simpletask5-archive'   # 归档加密盐值（结合 SECRET_KEY 派生 AES 密钥）
DJANGO_SIMPLETASK_ARCHIVE_RETENTION_DAYS = 7                    # 归档保留天数

Cron

DJANGO_SIMPLETASK_CRONJOB_AUTO_SYNC = True  # 是否自动将代码注册的 Cron 同步到数据库

安全

# 是否启用 Python/Shell 脚本执行器（默认关闭，开启有安全风险）
DJANGO_SIMPLETASK_ENABLE_SCRIPT_EXECUTORS = False

# 脚本执行白名单（空列表表示不限制）
DJANGO_SIMPLETASK_SCRIPT_WHITELIST = ['/path/to/allowed/scripts']

# 自定义字段加密（默认使用 django-safe-fields 的默认加密）
DJANGO_SIMPLETASK_FIELD_CIPHER_CLASS = None                              # 自定义加密器类
DJANGO_SIMPLETASK_RESULT_PASSWORD = None     # 执行结果加密密码
DJANGO_SIMPLETASK_ERROR_MSG_PASSWORD = None  # 错误信息加密密码
DJANGO_SIMPLETASK_CONTEXT_PASSWORD = None    # 上下文数据加密密码

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

快速开始

1. 定义任务模型

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',
    }

    simpletask_queue = {
        'create': 'django_simpletask5.queue.high_priority',
        'update': 'django_simpletask5.queue.default',
        'delete': 'django_simpletask5.queue.default',
    }

    trigger_update_fields = ['customer_name', 'amount', 'status']

2. 编写执行器

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

class OrderCreateExecutor(BaseExecutor):
    def execute(self, execution: TaskExecution) -> str | None:
        context = execution.get_context_dict()
        # 业务逻辑...
        return 'ok'

执行器可以通过 executor_class 属性覆盖超时配置：

class OrderCreateExecutor(BaseExecutor):
    timeout_seconds = 300                       # 执行超时 5 分钟
    pickup_timeout_seconds = 1800               # Pickup 超时 30 分钟

拒绝回退

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

from django_simpletask5.executors.exceptions import RejectException

class MyExecutor(BaseExecutor):
    def execute(self, execution: TaskExecution) -> 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 误回收

3. 启动 Worker

默认只监听 default 队列，high_priority 队列需要单独启动 Worker 处理。

# 启动 4 个 Worker 处理 default 队列
python manage.py django_simpletask_executor --workers 4

# 单独启动 Worker 处理 high_priority 队列
python manage.py django_simpletask_executor --queue django_simpletask5.queue.high_priority

# 指定执行器
python manage.py django_simpletask_executor --workers 2 --service myapp.executors.OrderCreateExecutor

4. 启动 Cron 调度

python manage.py django_simpletask_crontab

5. 触发自定义事件

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

Cron 任务

在代码中定义 Cron 任务

在任意 app 的 cronjobs.py 文件中使用 register_cronjob 注册：

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

register_cronjob(
    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='定期健康检查',
)

从代码同步到数据库

注册的 Cron 任务需要同步到数据库才会生效。框架默认在 django_simpletask_crontab 启动时自动同步（可通过 DJANGO_SIMPLETASK_CRONJOB_AUTO_SYNC = False 关闭），也可以手动执行：

python manage.py django_simpletask_sync_cronjobs

同步后，用户可以在 Django Admin 中查看和修改 Cron 任务，被手动修改过的任务不会在后续同步中被覆盖（is_modified_by_user 标记保护）。

内置执行器

执行器	说明
PingPongExecutor	健康检查，返回 'pong'
BashScriptExecutor	执行 Shell 脚本
PythonScriptExecutor	执行 Python 代码
SimpleRequestExecutor	发起 HTTP 请求
StatusCheckExecutor	检测卡住的执行并标记超时（每 5 分钟）
RetryTimeoutExecutor	重试超时的执行（每 10 分钟）
ArchiveExecutor	归档已完成执行并生成统计（每天凌晨 2 点）

架构

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

Releases

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