python - 如何在 Celery 任务中使用 Flask-SQLAlchemy

Question

我最近切换到 Celery 3.0。在此之前，我使用的是 Flask-Celery为了将 Celery 与 Flask 集成。虽然它有很多问题，比如隐藏一些强大的 Celery 功能，但它允许我使用 Flask 应用程序的完整上下文，尤其是 Flask-SQLAlchemy。

在我的后台任务中，我正在处理数据和 SQLAlchemy ORM 来存储数据。 Flask-Celery 的维护者已经放弃了对该插件的支持。该插件在任务中腌制 Flask 实例，因此我可以完全访问 SQLAlchemy。

我试图在我的 tasks.py 文件中复制此行为，但没有成功。您对如何实现这一点有任何提示吗？

Answer 1

更新:我们已经开始使用更好的方法来处理应用程序的拆卸和设置，基于描述的模式 in the more recent flask documentation .

extensions.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

一旦您以这种方式设置应用程序，您就可以运行和使用 celery，而无需在应用程序上下文中显式运行它，因为您的所有任务将在必要时自动在应用程序上下文中运行，而您不需要'不必明确担心任务后拆解，这是一个需要管理的重要问题(请参阅下面的其他回复)。

疑难解答

那些不断获得 with _celery.app.app_context(): AttributeError: 'FlaskCelery' object has no attribute 'app' 的人请确保:

1. 将 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)

1. 在你flask run和使用之前让你的 celery worker 开始工作

celery worker -A app:celery -l info -f celery.log

注意 app:celery，即从 app.py 加载。

您仍然可以从扩展导入来装饰任务，即 from extensions import celery。

下面的旧答案，仍然有效，但不是一个干净的解决方案

我更喜欢在应用程序上下文中运行所有 celery，方法是创建一个单独的文件，该文件在应用程序的上下文中调用 celery.start()。这意味着您的任务文件不必到处都是上下文设置和拆卸。它也很适合 flask “应用程序工厂”模式。

extensions.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()