Comment utiliser Flask-SQLAlchemy dans une tâche de céleri

mise à jour: nous avons depuis commencé à utiliser une meilleure façon de gérer l'application démonté et mis en place sur une base par tâche, sur la base du modèle décrit dans la documentation la plus récente de la fiole.

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

une fois que vous avez configuré votre application de cette façon, vous pouvez exécuter et utiliser celery sans avoir à l'exécuter explicitement dans un contexte d'application, car toutes vos tâches seront automatiquement exécutées dans un contexte de l'application si nécessaire, et vous n'avez pas à vous soucier explicitement du démontage post-tâche, qui est une question importante à gérer (voir les autres réponses ci-dessous).

Vieux-réponse ci-dessous, fonctionne toujours, mais pas aussi propre une solution

je préfère courir tous de céleri dans le contexte de l'application en créant un fichier séparé qui invoque le céleri.start () avec le contexte de l'application. Cela signifie que votre fichier de tâches n'a pas à être jonché de configuration de contexte et teardowns. Elle se prête également bien au modèle "application factory".

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