可以动态禁用 celery 调度程序任务吗？

Question

声明中是否有一个“启用”字段可以动态启用/禁用计划任务？

CELERYBEAT_SCHEDULE = {
    'add-every-30-seconds': {
        'task': 'tasks.add',
        'schedule': timedelta(seconds=30),
        'args': (16, 16)
        'enabled': env['PROD'] == 'True' <--- dynamically enable/disable task?
    },
}

否则我将不得不动态构造 dict 就像

if env['PROD'] == 'True':
    CELERYBEAT_SCHEDULE['add-every-30-seconds'] = {...}

enabled 字段看起来更干净

Answer 1

另一种方法是破解 schedule 字段，例如：

class NoSchedule(BaseSchedule):
    def is_due(self, last_run_at):
        return schedstate(is_due=False, next=99999999)

CELERYBEAT_SCHEDULE = {
    'add-every-30-seconds': {
        'task': 'tasks.add',
        'schedule': timedelta(seconds=30) if env['PROD'] == 'True' else NoSchedule(),
        'args': (16, 16)
    },
}