如何将每个基于 celery 类的任务保存在 django 项目中的单独文件中？

Question

我使用官方文档在我的 django 项目中设置了 celery
http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html#using-celery-with-django

所以我的项目结构是

└── mysite
    ├── db.sqlite3
    ├── manage.py
    ├── mysite
    │   ├── celery.py
    │   ├── __init__.py
    │   ├── settings.py
    │   ├── urls.py
    │   └── wsgi.py
    └── polls
        ├── admin.py
        ├── apps.py
        ├── forms.py
        ├── __init__.py
        ├── migrations
        │   ├── 0001_initial.py
        │   └── __init__.py
        ├── models.py
        ├── tasks.py
        ├── tests.py
        └── views.py  

polls

正在申请

polls/tasks.py

有基于课程的芹菜任务。
目前

tasks.py

有很多任务，因此文件太大。我想将每个任务保存在单独的文件中，例如

mysite/polls/
├── admin.py
├── apps.py
├── forms.py
├── __init__.py
├── migrations
│   ├── 0001_initial.py
│   └── __init__.py
├── models.py
├── tasks # I want to keep easy task in separate file like this
│   ├── __init__.py
│   ├── download_task.py
│   ├── process_task.py
│   └── upload_task.py
├── tests.py
└── views.py

如何使此设置发挥作用？

Answer 1

这是100%正确的。 在您的

tasks/__init__.py

文件中，确保从其他文件导入任务：

    from .download_task import *  
    from .process_task import *
    # etc...

然后确保您的 celery.py 文件中有

autodiscover_tasks

调用，以发现每个

INSTALLED_APPS

中的任务。

Answer 2

2023更新

您可以在 celery.py 中设置 autodiscover_tasks 来查找其他文件。

例如。 app.autodiscover_tasks(lated_name='tasksv2')

这将在您的应用程序文件夹中搜索“taskv2.py”。

Answer 3

在实例化 Celery 应用程序的文件中。

试试这个：

def make_celery():
    celery = Celery()
    celery.conf.broker_url = os.environ.get(
        "CELERY_BROKER_URL", "redis://localhost:6379"
    )
    celery.conf.result_backend = os.environ.get(
        "CELERY_RESULT_BACKEND", "redis://localhost:6379"
    )
    return celery

celery = make_celery()

# for class based task
# from tasks.download_task import DownloadTask 
celery.register_task(DownloadTask)

# for function based task
# from tasks.download_task import download_task
celery.task(download_task)