Django-Q2 实现模型方法定时任务方法

在 Django-Q2 中，直接将模型实例方法（如 self.run_function）作为定时任务的 func 或 hook 会导致序列化失败，因其绑定方法无法跨进程反序列化；正确做法是使用字符串路径引用独立函数，并通过 kwargs 传递实例 ID 进行上下文重建。

在 Django-Q2 中，直接将模型实例方法（如 `self.run_function`）作为定时任务的 `func` 或 `hook` 会导致序列化失败，因其绑定方法无法跨进程反序列化；正确做法是使用字符串路径引用独立函数，并通过 `kwargs` 传递实例 ID 进行上下文重建。

Django-Q2 的 Schedule 机制要求任务函数必须是可导入的、无状态的模块级函数，即以 'package.module.function_name' 形式的点分字符串（dotted string） 表示。而 self.run_function 是一个绑定到具体模型实例的 bound method，在任务被 worker 进程反序列化时，该实例已不存在，Python 无法重建其 self 上下文，因此抛出 Function is not defined 错误。

相比之下，async_task() 在当前进程上下文中立即序列化并入队，可能仍持有对活动对象的弱引用或临时状态，故偶发成功——但这属于未定义行为，不可靠且不适用于持久化定时任务。

✅ 正确解决方案：将逻辑解耦至 tasks.py，使用字符串路径注册任务，并通过 kwargs 传递关键参数（如模型 ID）：

# models.py
from django.db import models

class TestApp(models.Model):
    cron = models.CharField(max_length=200)
    args = models.CharField(max_length=200)
    test_function = models.ForeignKey('TestFunction', on_delete=models.CASCADE)
    scheduled_task = models.ForeignKey(
        'django_q.Schedule',
        blank=True,
        null=True,
        on_delete=models.SET_NULL,
        related_name='+'
    )

    def get_args(self):
        import ast
        return ast.literal_eval(self.args)

    def save(self, *args, **kwargs):
        # 删除旧调度（可选）
        if self.scheduled_task:
            self.scheduled_task.delete()

        # 创建新 Schedule，使用字符串路径 + 实例 ID
        self.scheduled_task = Schedule.objects.create(
            func='test_app.tasks.run_function',      # ✅ 模块级函数路径
            hook='test_app.tasks.print_task',       # ✅ 同样为字符串路径
            schedule_type=Schedule.CRON,
            cron=self.cron,
            kwargs={'TestApp_id': self.id}          # ✅ 传递唯一标识，供任务内重建实例
        )
        super().save(*args, **kwargs)

# tasks.py（位于 test_app/ 目录下）
from test_app.models import TestApp

def run_function(**kwargs):
    """执行实际业务逻辑"""
    test_app_id = kwargs['TestApp_id']
    try:
        test_app = TestApp.objects.get(pk=test_app_id)
    except TestApp.DoesNotExist:
        raise RuntimeError(f"TestApp with id {test_app_id} not found")

    # 动态导入并调用目标函数
    import ast
    import importlib
    module = importlib.import_module(test_app.test_function.library_name)
    func = getattr(module, test_app.test_function.function_name)

    result = func(*test_app.get_args())
    print(f"[Task] Executed for TestApp {test_app_id}: {result}")

    # 返回结构化结果，供 hook 使用
    return {
        'result': result,
        'TestApp_id': test_app_id
    }

def print_task(task):
    """Hook 函数：接收 task 对象，提取 run_function 的返回值"""
    if task.success and isinstance(task.result, dict):
        app_id = task.result.get('TestApp_id')
        result_val = task.result.get('result')
        print(f"[Hook] TestApp {app_id} completed → {result_val}")
    else:
        print(f"[Hook] Task {task.name} failed or returned invalid result")

⚠️ 注意事项：

• tasks.py 必须位于 Django 可识别的应用目录中（如 test_app/tasks.py），且模块路径需与 INSTALLED_APPS 中注册的应用名一致；
• hook 函数签名固定为 hook(task)，其中 task 是 django_q.models.Task 实例，其 result 属性即 func 的返回值；
• 避免在 run_function 中直接操作未保存的模型实例或依赖请求上下文（如 request）——定时任务无 HTTP 上下文；
• 建议为 run_function 添加异常捕获与日志记录，防止单个失败任务阻塞整个调度队列；
• 若需更高可靠性，可在 run_function 开头加锁（如基于 TestApp.id 的 Redis 锁），防止同一实例被并发触发。

通过这种解耦设计，既满足 Django-Q2 对函数可序列化的要求，又保持了业务逻辑与模型职责的清晰分离，是生产环境中推荐的标准实践。