最近、Celery 3.0 に切り替えました。その前は、Celery を Flask と統合するためにFlask-Celeryを使用していました。Celery の強力な機能がいくつか隠されているなど、多くの問題がありましたが、Flask アプリ、特に Flask-SQLAlchemy の完全なコンテキストを使用することができました。
私のバックグラウンド タスクでは、データと SQLAlchemy ORM を処理してデータを保存しています。Flask-Celery のメンテナーはプラグインのサポートを終了しました。プラグインはタスクで Flask インスタンスをピクルしていたので、SQLAlchemy に完全にアクセスできました。
tasks.py ファイルでこの動作を再現しようとしていますが、成功していません。これを達成する方法についてのヒントはありますか？
extends.py
import flask
from flask.ext.sqlalchemy import SQLAlchemy
from celery import Celery
class FlaskCelery(Celery):
def __init__(self, *args, **kwargs):
super(FlaskCelery, self).__init__(*args, **kwargs)
self.patch_task()
if 'app' in kwargs:
self.init_app(kwargs['app'])
def patch_task(self):
TaskBase = self.Task
_celery = self
class ContextTask(TaskBase):
abstract = True
def __call__(self, *args, **kwargs):
if flask.has_app_context():
return TaskBase.__call__(self, *args, **kwargs)
else:
with _celery.app.app_context():
return TaskBase.__call__(self, *args, **kwargs)
self.Task = ContextTask
def init_app(self, app):
self.app = app
self.config_from_object(app.config)
celery = FlaskCelery()
db = SQLAlchemy()
app.py
from flask import Flask
from extensions import celery, db
def create_app():
app = Flask()
#configure/initialize all your extensions
db.init_app(app)
celery.init_app(app)
return app
この方法でアプリを設定すると、必要に応じてすべてのタスクがアプリケーションコンテキストで自動的に実行されるため、アプリケーションコンテキスト内から明示的に実行しなくても、セロリを実行して使用できます。管理する重要な問題であるタスク後の分解について明示的に心配する（以下の他の応答を参照）。
取得し続ける人は、次のwith _celery.app.app_context(): AttributeError: 'FlaskCelery' object has no attribute 'app'ことを確認してください。
celeryインポートはapp.pyファイルレベルで保持します。避ける：app.py
from flask import Flask
def create_app():
app = Flask()
initiliaze_extensions(app)
return app
def initiliaze_extensions(app):
from extensions import celery, db # DOOMED! Keep celery import at the FILE level
db.init_app(app)
celery.init_app(app)
flask run、使用してくださいcelery worker -A app:celery -l info -f celery.log
app:celery、つまりからの読み込みに注意してくださいapp.py。
拡張機能からインポートして、タスクを装飾することもできますfrom extensions import celery。
私は、アプリケーションのコンテキストでcelery.start（）を呼び出す別のファイルを作成することにより、アプリケーションコンテキスト内ですべてのセロリを実行することを好みます。これは、タスクファイルにコンテキストの設定や分解を散らかす必要がないことを意味します。また、フラスコの「アプリケーションファクトリ」パターンにも適しています。
extends.py
from from flask.ext.sqlalchemy import SQLAlchemy
from celery import Celery
db = SQLAlchemy()
celery = Celery()
tasks.py
from extensions import celery, db
from flask.globals import current_app
from celery.signals import task_postrun
@celery.task
def do_some_stuff():
current_app.logger.info("I have the application context")
#you can now use the db object from extensions
@task_postrun.connect
def close_session(*args, **kwargs):
# Flask SQLAlchemy will automatically create new sessions for you from
# a scoped session factory, given that we are maintaining the same app
# context, this ensures tasks have a fresh session (e.g. session errors
# won't propagate across tasks)
db.session.remove()
app.py
from extensions import celery, db
def create_app():
app = Flask()
#configure/initialize all your extensions
db.init_app(app)
celery.config_from_object(app.config)
return app
RunCelery.py
from app import create_app
from extensions import celery
app = create_app()
if __name__ == '__main__':
with app.app_context():
celery.start()
tasks.py ファイルで次の操作を行います。
from main import create_app
app = create_app()
celery = Celery(__name__)
celery.add_defaults(lambda: app.config)
@celery.task
def create_facet(project_id, **kwargs):
with app.test_request_context():
# your code