python - Flask with Celery - 应用程序上下文不可用

Question

我有一个 Flask 应用程序，注册如下:

APP = Flask(__name__)
APP.config.from_object('config')

我已经为 URL 定义了一个 View ，其中调用了一个与数据库交互的函数。

from tasks import some_func
.
.
.
some_func.delay(params)

在 tasks.py 文件中，我正在创建一个 Celery 实例，如下所示:

# Config values don't work here
celery = Celery('tasks', broker='amqp://', backend='amqp://')
.
.
.
@celery.task()
def some_func():
    #DB interactions

现在我收到一条错误消息:

RuntimeError: Working outside of application context.

我阅读了应用程序上下文以及如何使用它们。我在我的 tasks.py 文件中导入了 current_app 并尝试使用如下上下文:

@celery.task()
def some_func():
    with current_app.app_context():
        #DB interactions

但是我仍然得到同样的错误。我还尝试按如下方式从主文件推送上下文:

ctx = APP.app_context()
ctx.push()

但还没有运气。

如何让 Celery 与 Flask 一起工作？

注意:我已经尝试过他们的 example here .

Answer 1

上下文:尝试通过 Celery 任务发送电子邮件时，同样的错误发生在我身上。就我而言，我正在发送由 Flask 使用 html 模板呈现的电子邮件。

原因:Celery 和 Flask 断开连接。因此，Celery 对 Flask 的应用上下文一无所知。我不得不手动注入(inject)应用程序上下文。

解决方案:对我有用的是 Flask 的 app_context() 函数。只需在错误源自的函数周围注入(inject)/包装应用程序上下文。

示例:

from app import app
from app import mail

@celery.task
def sign_up_task(**config, **context):
    mail = Mail()
    with app.app_context():
        mail.send(**config, **context)