Django-q2 类函数上的调度任务挂钩

Question

我有一个这样的对象：

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(Schedule, blank=True, null=True, on_delete=models.SET_NULL)

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

    def run_function(self):
        Module = __import__(self.test_function.library_name)
        func = getattr(Module, self.test_function.function_name)
        result = func(*self.get_args())
        print(result)
        return result

    def print_task(self, task):
        print(self.id, task)

我对这样的计划任务感兴趣：

def save(self, *args, **kwargs):
    self.scheduled_task = Schedule.objects.create(
        func=self.run_function,
        hook=self.print_task,
        schedule_type=Schedule.CRON,
        cron=self.cron
    )
    super(TestApp, self).save(*args, **kwargs)

但这不起作用，会导致以下结果：

18:32:02 [Q] INFO Process-f625bf1f4a024df8be5e15647bf294a9 created task alaska-ten-october-mirror from schedule [4]
18:32:02 [Q] INFO Process-069b6ca530ae4e83be6aedbd669a94a7 processing alaska-ten-october-mirror '<bound method TestApp.run_function of <TestApp: TestApp object (1)>>' [4]
18:32:02 [Q] ERROR malformed return hook '<bound method TestApp.print_task of <TestApp: TestApp object (1)>>' for [alaska-ten-october-mirror]
18:32:02 [Q] ERROR Failed '<bound method TestApp.run_function of <TestApp: TestApp object (1)>>' (alaska-ten-october-mirror) - Function <bound method TestApp.run_function of <TestApp: TestApp object (1)>> is not defined : Traceback (most recent call last):

当我进行简单的异步调用时，它会起作用：

def save(self, *args, **kwargs):
    async_task(
        func=self.run_function,
        hook=self.print_task
    )
    super(TestApp, self).save(*args, **kwargs)

这将正确完成工作：

18:35:25 [Q] INFO Process-cfc73b4a7c5d48d69eed82b311f18250 processing ceiling-echo-six-west '<bound method TestApp.run_function of <TestApp: TestApp object (1)>>'
-2.0
1 ceiling-echo-six-west

我不知道为什么异步可以做到但时间表不行。

Answer 1

我想这是因为计划任务需要能够独立于其他项目运行，但上面的问题是因为：

• 任务可以有一个引用函数，如函数和钩子
• 时间表需要将功能设置为“仅限点线”。

如何解决：

def save(self, *args, **kwargs):
    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(TestApp, self).save(*args, **kwargs)

然后在你的tasks.py中我这样定义了任务：

def run_function(**kwargs):
    id = kwargs['TestApp_id']
    test_app = TestApp.objects.get(pk=id)
    Module = __import__(test_app.test_function.library_name)
    func = getattr(Module, test_app.test_function.function_name)
    result = func(*test_app.get_args())
    print(result)
    return {
        'result': result,
        'TestApp_id': id
    }


def print_task(task):
    print(task.result['TestApp_id'], task.result['result'])