Celery在包含其他文件名的文件中发现任务

Question

我开始有这么多的芹菜任务,我想将它们分解成较小的文件而不是将它们放在一个大的文件中tasks.py,但是我无法让芹菜去发现它们.

工作结构:

proj/
    app/
        tasks.py

结构我想工作:

proj/
    app/
        tasks/
            __init__.py
            task1.py
            task2.py
            ...

但这里芹菜找不到我的任务.我尝试设置CELERY_IMPORTS,但后来我必须指定每个文件,并且文件太多了.如果我导入每个文件的内容为相同__init__.py.我希望自动发现所有任务,或至少任务目录中的文件中的所有任务.

我当然可以将每个任务目录中的所有文件列入CELERY_IMPORTS,但这看起来相当丑陋.

任何想法,如果这是可能的一个很好的方式？

Answer 1

你可以__all__在你的tasks/__init__.py文件中定义。没有额外的芹菜设置更改。

例如：

# app/tasks/__init__.py

from .file_with_tasks_1 import task1, task2
from .file_with_tasks_2 import task3

__all__ = [
    'task1',
    'task2',
    'task3',
]

在 celery v4.4.6 上测试

Answer 2

我已经放弃寻找一个好的解决方案，而是编写了一个我调用并填充 CELERY_IMPORTS 的函数。这不太好，但是很有效。

这是供将来参考的代码：

import os

def detect_tasks(project_root):
    tasks = []
    file_path = os.path.join(project_root, 'apps')
    for root, dirs, files in os.walk(file_path):
        for filename in files:
            if os.path.basename(root) == 'tasks':
                if filename != '__init__.py' and filename.endswith('.py'):
                    task = os.path.join(root, filename)\
                        .replace(os.path.dirname(project_root) + '/', '')\
                        .replace('/', '.')\
                        .replace('.py', '')
                    tasks.append(task)
    return tuple(tasks)

然后在设置中：

CELERY_IMPORTS = detect_tasks(project_root)

其中project_root 会是这样的：

project_root = os.path.dirname(os.path.abspath(__file__))

Answer 3

celery 默认搜索 tasks.py 的唯一原因是 autodiscover_tasks 的默认参数：

./loaders/base.py:def autodiscover_tasks(packages, related_name='tasks'):

如果您使用文档推荐的配置，您可以只调用 autodiscover_tasks，并为您希望拥有任务的不同文件名使用 related_name 的非默认值。例如这里是我们的 celery.py：

from __future__ import absolute_import

import os

from celery import Celery

# set the default Django settings module for the 'celery' program.
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "settings")

from django.conf import settings

app = Celery('app')

# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS, related_name='tasks2')