python - 使用 Flask 和依赖注入(inject)运行 Celery

Question

我有一个使用依赖注入(inject)和 celery 的 Flask 应用程序。我有下面的示例，但除了创建第二个 Flask 实例之外，我的主应用程序还必须实例化 celery 所需的所有模块。有没有更好的方法来实现这一点？

特别是:

为什么主“前端”应用程序应该依赖整个 Celery“后端”堆栈来配置 Celery 客户端？我想解耦这些子系统，因为前端只是启动任务。

主要.py

import tasks.py
app = Flask(__name__)
FlaskInjector(app=app, modules=[A, B, C, D, E, F])

celery .py

app = Flask(__name__)
injector = Injector(modules=[A, B])
FlaskInjector(app=app, injector=injector)
celery = Celery(app.import_name, include=['tasks'])

任务.py

from celery import celery, injector
@celery.task
def my_task():
    injector.get(A).foo()

我不从 main 导入 app 因为我不希望 Celery 依赖主应用程序中与运行无关的所有内容任务。相反，我不希望我的主应用程序依赖于配置工作人员所需的 Celery 客户端的所有 Bootstrap 。这对于玩具应用程序来说很好，但随着大型系统的增长，管理这些依赖关系很重要，我不明白 Celery 客户端配置(调用任务)的分离以及工作人员需要什么。

我有一个前端 Flask 应用程序和一个后端 Celery 应用程序。 Celery 应用程序具有我不希望 Flask 依赖的业务逻辑。它很大，很复杂，而且经常变化。我不想让 Flask 应用程序膨胀，每次 Celery 更改时都重新部署它，或者让我的 Flask 开发人员接触 Celery 代码。但据我所知，任务的 Flask 调用不能与它们的 Celery 实现分离。

Answer 1

这是答案(由@kindule 友情提供):

app.send_task('my_task', ...)

http://docs.celeryproject.org/en/latest/reference/celery.html?highlight=send_task#celery.Celery.send_task

https://github.com/celery/celery/issues/3474#issuecomment-249723398