背景
最近写 Django
时遇到个问题,就是要使用 Celery
进行定时任务触发。在配置任务函数时老是无法自动发现注册过的任务。最终花了点心思,研究了下源码才解决。在这里记录下注意点
在使用 Celery
提供的 autodiscover_tasks
方法来自动发现任务时,默认是发现在 settings
模块里注册过的应用下的 tasks
模块中包含的被 Celery
的 task()
装饰器装饰过的任务函数。
而我想将这些任务函数分开放在不同的模块下,按照包的方式进行管理。这里就存在三个坑,需要注意下。
注意点
- 为了尽量减少对
Celery
默认逻辑的修改,包名应该为tasks
。 - 存在
__init__
模块的目录才是Python
包。记得创建一个空的__init__.py
文件。 - 包中定义的任务,都需要在
__init__
模块中进行导入,最好将这些被导入的任务函数包含在__all__
列表中,来避免被flake8
扫出警告。