我在用python-flask应用程序工厂应用程序实现芹菜时遇到问题
我打算从应用程序init文件创建一个Celery应用程序的实例，如下所示：
from celery import Celery
celery = Celery('myapp', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')
当调用时，我不能使用其他蓝图中的Celery。
def init_celery(app):
celery = Celery()
celery.conf.broker_url = app.config['CELERY_BROKER_URL']
celery.conf.result_backend = app.config['CELERY_RESULT_BACKEND']
celery.conf.update(app.config)
class ContextTask(celery.Task):
"""Make celery tasks work with Flask app context"""
def __call__(self, *args, **kwargs):
with app.app_context():
return self.run(*args, **kwargs)
celery.Task = ContextTask
return celery
当create_app:时初始化芹菜
init_celery(app)
了解芹菜是如何在这个烧瓶炊具中实现的
Joost Döbken的答案可能有效，但似乎比实际情况复杂一些。
我发现Miguel Grinberg的更简单的解决方案对我来说很好：
from celery import Celery
from config import config, Config
celery = Celery(__name__, broker=Config.CELERY_BROKER_URL)
def create_app(config_name):
# ...
celery.conf.update(app.config)
# ...
return app
https://blog.miguelgrinberg.com/post/celery-and-the-flask-application-factory-pattern
API的回答是正确的。。。还要将以下内容添加到芹菜配置中。。。这将有助于防止在broker关闭或无法联系到时，由芹菜引起的无休止重试。。
broker_transport_options = {
'max_retries': 3,
'interval_start': 0,
'interval_step': 0.2,
'interval_max': 0.5,
}
在回答这个问题时，使用的芹菜版本为==4.3。
flask执行插件支持开箱即用的应用程序工厂模式：
from flask import Flask
from flask_execute import Celery
celery = Celery()
app = Flask(__name__)
celery.init_app(app)