如何正确指定Django+Celery Beat周期性任务的路由？

问题场景 #

已完成Django项目的Celery基础配置，尝试创建周期性任务时出现KeyError: 'apps.laws.tasks.periodic.SampleTask'错误。

项目结构 #

project
├── apps
│   ├── laws
│        └──tasks
│           └──periodic.py # 任务定义文件
├── config
│   ├── celery.py
│   ├── settings
│      └── base.py # 配置CELERY_BEAT_SCHEDULE

当前配置与代码 #

1. base.py中的定时任务配置

CELERY_BEAT_SCHEDULE = {
    "sample_task": {
        "task": "apps.laws.tasks.periodic.SampleTask", # 错误所在行
        "schedule": crontab(minute="*/1"),
    },
}

2. periodic.py中的任务类

class SampleTask(Task):
    name="laws.sample_task"

    def run(self, operation, *args, **kwargs):
        logger.info("The sample task in running...")

3. 报错信息

The delivery info for this task is:
 {'exchange': '', 'routing_key': 'celery'}
 Traceback (most recent call last):
   File "/usr/local/lib/python3.9/site-packages/celery/worker/consumer/consumer.py", line 591, in on_task_received
     strategy = strategies[type_]
 KeyError: 'apps.laws.tasks.periodic.SampleTask'

修复步骤 #

1. 修正定时任务的task字段值 #

Celery的定时任务配置中，task字段需要填写任务类定义的name属性值，而非类的导入路径。修改base.py的配置：

CELERY_BEAT_SCHEDULE = {
    "sample_task": {
        "task": "laws.sample_task", # 改为任务类的name属性值
        "schedule": crontab(minute="*/1"),
    },
}

2. 确保Celery Worker能发现任务 #

需要让Celery worker识别到这个任务类，有两种常用方式：

• 手动导入任务：在config/celery.py中添加任务导入语句

from apps.laws.tasks.periodic import SampleTask

• 自动发现任务：如果已开启app.autodiscover_tasks()，在apps/laws/apps.py的AppConfig类中添加任务导入逻辑

from django.apps import AppConfig

class LawsConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'apps.laws'

    def ready(self):
        import apps.laws.tasks.periodic

3. 修复任务方法的参数问题 #

当前run方法要求传入operation参数，但定时任务默认不会传递该参数，会导致后续执行报错。给参数设置默认值：

def run(self, operation=None, *args, **kwargs):
    logger.info("The sample task in running...")

验证 #

重启Celery beat和worker进程，观察任务是否正常执行。

内容的提问来源于stack exchange，提问作者Mahdi Jafari