我有一个烧瓶应用程序,该应用程序注册如下:
APP = Flask(__name__)
APP.config.from_object('config')
我已经定义了称为函数的URL的视图,该函数与DB相互作用。
from tasks import some_func
.
.
.
some_func.delay(params)
在tasks.py文件中,我正在创建一个芹菜实例,如下所示:
# 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
使用HTML模板发送的。
原因: Celery
和Flask
断开连接。因此,芹菜对烧瓶的应用上下文一无所知。我必须手动注入应用程序上下文。
解决方案:对我有用的是烧瓶的app_context()
功能。只需将误差起源的函数注入/包装应用程序上下文。
示例:
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)