为什么芹菜自动发现不映射来自不同文件夹的任务？

Question

我有以下芹菜项目的结构，它不需要使用django/flask，但我正试图根据此站点的芹菜结构布局映射我的任务https://docs.celeryproject.org/en/latest/_modules/celery/app/base.html#Celery.autodiscover_tasks我可能会遇到一些自动发现问题，但我仍然没有从不同文件夹映射的任务

.
├── docker-compose.yml
├── LICENSE
├── logs
├── README.md
├── redis
│   ├── Dockerfile
│   └── redis.conf
└── thirdparty
    ├── appone
    │   ├── __init__.py
    │   └── tasks.py
    ├── apptwo
    │   ├── __init__.py
    │   └── tasks.py
    └── celery_tasks
        ├── celery.py
        ├── __init__.py
        └── settings.py

第三方/Celly\u任务/Celly.py

from celery import Celery

app = Celery("thirdparty")
app.config_from_object('celery_tasks.settings', namespace="CELERY")
app.autodiscover_tasks(['celery_tasks.appone', 'celery_tasks.apptwo'])

第三方/芹菜任务/settings.py

from celery.schedules import crontab
from datetime import timedelta


CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/0"
CELERY_BEAT_SCHEDULE = {
    "sample_task": {
        "task": "appone.tasks.add",
        'schedule': timedelta(seconds=30),
    },
    "send_email_report": {
        "task": "apptwo.tasks.mult",
        "schedule": crontab(minute="*"),
    },
}

# run task every 30 minutes
CELERY_BEAT_INTERVAL = 30 * 60

第三方/apptwo/tasks.py

import time
from celery_tasks.celery import app

@app.task
def mult(x, y):
    print('start apptwo mult function')
    time.sleep(10)
    print('result:', x * y)
    return x * y

Answer 1

让我们关注这棵路径树：

└── thirdparty
    ├── appone
    │   ├── __init__.py
    │   └── tasks.py
    ├── apptwo
    │   ├── __init__.py
    │   └── tasks.py
    └── celery_tasks
        ├── celery.py
        ├── __init__.py
        └── settings.py

相对于外部项目（作为源根），模块列表如下：

thirdparty.appone.tasks (contains add)
thirdparty.apptwo.tasks (contains mult)
thirdparty.celery_tasks.celery (contains app)
thirdparty.celery_tasks.settings (contains CELERY_BROKER_URL, CELERY_RESULT_BACKEND, etc.)

相对于第三方（作为源根），模块列表如下：

appone.tasks (contains add)
apptwo.tasks (contains mult)
celery_tasks.celery (contains app)
celery_tasks.settings (contains CELERY_BROKER_URL, CELERY_RESULT_BACKEND, etc.)

根据您从thirdparty/apptwo/tasks.py导入芹菜应用程序的方式，我们可以说您已将第三方设计为根源：

from celery_tasks.celery import app  # If the sources root was the outer project, then this would be -> from thirdparty.celery_tasks.celery import app

您访问./thirdparty/appone/和./thirdparty/apptwo/的方式错误：

app.autodiscover_tasks(['celery_tasks.appone', 'celery_tasks.apptwo'])

为什么?？芹菜任务模块没有.appone或.apptwo。它只有。芹菜和。设置。如果您查看我上面描述的模块列表，这是正确的路径：

app.autodiscover_tasks(['appone', 'apptwo'])  # This assumes that thirdparty is your sources root, thus you wouldn't write it as thirdparty.appone nor thirdparty.apptwo

进一步阅读

考虑到thirdparty是源根，您必须将导入的格式设置为始终相对于它，例如from celery_tasks.celery import app，而不是相对于外部项目，例如^{或例如from myproj.thirdparty.celery_tasks.celery import app

然后，您必须告诉PYTHONPATH关于它的信息，以便可以执行相对于第三方目录的导入（例如您如何使用from celery_tasks.celery import app）

export PYTHONPATH=${PYTHONPATH}:/path/to/myproj/thirdparty

为什么?？假设您刚刚定义了PYTHONPATH到外部项目：

export PYTHONPATH=${PYTHONPATH}:/path/to/myproj

然后，您必须导入相对于myproj而不是相对于第三方的模块。因此，在thirdparty/apptwo/tasks.py中导入的方式是：

from thirdparty.celery_tasks.celery import app

而您的自动发现可能是：

app.autodiscover_tasks(['thirdparty.appone', 'thirdparty.apptwo'])