Django-celery-beat动态定时任务配置报CrontabSchedule赋值错误

Cannot assign "(<CrontabSchedule: 0 0 27 1 * (m/h/dM/MY/d) UTC>, True)": "PeriodicTask.crontab" must be a "CrontabSchedule" instance.

这个报错是字段赋值类型不匹配导致的：

• Django ORM的get_or_create()方法返回值是长度为2的元组，格式为(查询到/新创建的模型实例, 标记是否为本次新建的布尔值)
• 你定义的crontab_schedule属性直接把get_or_create()的返回值原样返回，实际传给crontab字段的是包含CrontabSchedule实例和布尔值的元组，不是纯CrontabSchedule实例，自然触发类型校验失败。

除了核心报错外，现有代码还存在3个会导致运行异常的隐患：

1. Celery任务中写的Compliance.object.get存在笔误，Django模型管理器的固定名称是objects，漏写s会触发属性不存在错误
2. PeriodicTask.name字段要求全局唯一，直接用Rule的name字段赋值容易出现重名冲突，触发数据库唯一键报错
3. crontab_schedule属性最后抛出异常的代码里，对CharField类型的compliance_frequency字段调用了.value属性，字符串没有这个属性，触发不支持的频率时会直接报错
4. 创建CrontabSchedule时未指定时区，默认使用UTC时区，和国内业务时区差8小时，会导致定时任务执行时间和预期不符

• 核心修复：修改crontab_schedule属性，拆解get_or_create()的返回值，只返回CrontabSchedule实例
• 修正笔误：把Celery任务里的Compliance.object改成Compliance.objects，移除CharField字段不存在的.value调用
• 规避重名：给PeriodicTask的name字段拼接唯一的rule_id后缀，保证全局唯一
• 时区对齐：创建CrontabSchedule时传入和业务一致的时区参数，避免执行时间偏移

1. Rule模型部分 #

import json
from django.utils import timezone
import uuid
from django.db import models
from django_celery_beat.models import CrontabSchedule, PeriodicTask


class Rule(models.Model):
    rule_id = models.UUIDField(
         primary_key = False,
         default = uuid.uuid4,
         editable = True)
    name = models.CharField('Rule Name',max_length=255,blank=True,null=True)
    compliance_frequency = models.CharField(max_length=40,choices=(('QUARTERLY','QUARTERLY'),
    ('MONTHLY','MONTHLY'),('YEARLY','YEARLY')),default='YEARLY')
    applicable_from = models.DateField(default=timezone.now)
    applicable_to = models.DateField(null=True,blank=True)
    task = models.OneToOneField(
        PeriodicTask,
        on_delete=models.CASCADE,
        null=True,
        blank=True
    )
       
    def setup_task(self):
        self.task = PeriodicTask.objects.create(
            # 拼接rule_id保证任务名全局唯一
            name=f"{self.name}_{self.rule_id}",
            task='create_compliance',
            crontab=self.crontab_schedule,
            args=json.dumps([str(self.id),self.name,self.compliance_frequency]),
            start_time= timezone.now(),
            enabled=True
        )
        self.save()

    @property
    def crontab_schedule(self):
        # 国内业务可在参数中加 timezone='Asia/Shanghai' 对齐东八区时间
        if self.compliance_frequency == "QUARTERLY":
            schedule, _ = CrontabSchedule.objects.get_or_create(
                minute='0',
                hour='0',
                day_of_week='*',
                day_of_month='27',
                month_of_year='3,6,9,12'
            )
            return schedule
        if self.compliance_frequency == "MONTHLY":
            schedule, _ = CrontabSchedule.objects.get_or_create(
                minute='0',
                hour='0',
                day_of_week='*',
                day_of_month='27',
                month_of_year='*'
            )
            return schedule
        if self.compliance_frequency == 'YEARLY':
            schedule, _ = CrontabSchedule.objects.get_or_create(
                minute='0',
                hour='0',
                day_of_week='*',
                day_of_month='27',
                month_of_year='1'
            )
            return schedule

        raise NotImplementedError(
            '''Interval Schedule for {interval} is not added.'''.format(
                interval=self.compliance_frequency))

2. 信号配置部分 #

from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Rule


@receiver(post_save,sender=Rule)
def create_periodic_task(sender, instance, created, **kwargs):
    if created:
        instance.setup_task()
        print('signal working')

3. Celery任务部分 #

from __future__ import absolute_import, unicode_literals
from celery.utils.log import get_task_logger
from celery import shared_task
from test3.models import Compliance

logger = get_task_logger(__name__)


@shared_task
def create_compliance(id,name,compliance_frequency):
    compliance = Compliance.objects.create(rule_id=id,name=name,compliance_frequency=compliance_frequency)
    logger.info(f"periodic task run success, created compliance id: {compliance.id}")

内容的提问来源于stack exchange，提问作者Practical Bong