Flask with create_app, SQLAlchemy and Celery

Question

I'm really struggeling to the get the proper setup for Flask, SQLAlchemy and Celery. I have searched extensively and tried different approaches, nothing really seems to work. Either I misse the application context or can't run the workers or some other problems. The structure is very general so that I can build a bigger application.

I'm using: Flask 0.10.1, SQLAlchemy 1.0, Celery 3.1.13, my current setup is the following:

app/__init__.py

#Empty

app/config.py

import os
basedir = os.path.abspath(os.path.dirname(__file__))

class Config:

    @staticmethod
    def init_app(app):
        pass

class LocalConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = r"sqlite:///" + os.path.join(basedir, 
                                 "data-dev.sqlite")
    CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'


config = {
    "local": LocalConfig}

app/exstensions.py

from flask.ext.sqlalchemy import SQLAlchemy
from celery import Celery

db = SQLAlchemy()
celery = Celery()

app/factory.py

from extensions import db, celery
from flask import Flask
from flask import g
from config import config

def create_before_request(app):
    def before_request():
        g.db = db
    return before_request


def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])

    db.init_app(app)
    celery.config_from_object(config)

    # Register the blueprints

    # Add the before request handler
    app.before_request(create_before_request(app))
    return app

app/manage.py

from factory import create_app

app = create_app("local")

from flask import render_template
from flask import request

@app.route('/test', methods=['POST'])
def task_simple():
    import tasks
    tasks.do_some_stuff.delay()
    return ""

if __name__ == "__main__":
    app.run()

app/models.py

from extensions import db

class User(db.Model):
    __tablename__ = "user"

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(128), unique=True, nullable=False)

app/tasks.py

from extensions import celery
from celery.signals import task_prerun
from flask import g, current_app


@task_prerun.connect
def close_session(*args, **kwargs):
    with current_app.app_context():
       # use g.db
       print g

@celery.task()
def do_some_stuff():
    with current_app.app_context():
       # use g.db
       print g

in the folder app:

starting the development webserver with: python.exe manage.py
starting the workers with: celery.exe worker -A tasks

I get an import error that doesn't make any sense to me. Should I structure the application differently? At the end I think I want a quite basic setup, e.g. Using Flask with the factory pattern, be able to use the Flask-SQLAlchmey extension and have some worker that needs to access the database.

Any help is highly appreciated.

Traceback is when starting the celery worker.

Traceback (most recent call last):

  File "[PATH]\scripts\celery-script.py", line 9, in <module>
    load_entry_point('celery==3.1.13', 'console_scripts', 'celery')()

  File "[PATH]\lib\site-packages\celery\__main__.py", line 30, in main
    main()

  File "[PATH]\lib\site-packages\celery\bin\celery.py", line 81, in main
    cmd.execute_from_commandline(argv)

  File "[PATH]\lib\site-packages\celery\bin\celery.py", line 769, in execute_from_commandline
    super(CeleryCommand, self).execute_from_commandline(argv)))

  File "[PATH]\lib\site-packages\celery\bin\base.py", line 305, in execute_from_commandline
    argv = self.setup_app_from_commandline(argv)

  File "[PATH]\lib\site-packages\celery\bin\base.py", line 473, in setup_app_from_commandline
    user_preload = tuple(self.app.user_options['preload'] or ())
AttributeError: 'Flask' object has no attribute 'user_options'

UPDATE I change the code according to the suggestion in the comment. The worker starts now up but when test it with a get request to http://127.0.0.1:5000/test. I get the following traceback:

Traceback (most recent call last):
  File "[PATH]\lib\site-packages\celery\app\trace.py", line 230, in trace_task
    args=args, kwargs=kwargs)

  File "[PATH]\lib\site-packages\celery\utils\dispatch\signal.py", line 166, in send
    response = receiver(signal=self, sender=sender, \**named)

  File "[PATH]\app\stackoverflow\tasks.py", line 7, in close_session
    with current_app.app_context():

  File "[PATH]\lib\site-packages\werkzeug\local.py", line 338, in __getattr__
    return getattr(self._get_current_object(), name)

  File "[PATH]\lib\site-packages\werkzeug\local.py", line 297, in _get_current_object
    return self.__local()

  File "[PATH]\lib\site-packages\flask\globals.py", line 34, in _find_app
    raise RuntimeError('working outside of application context')
RuntimeError: working outside of application context exc, exc_info.traceback)))

UPDATE Based on the comments from Marteen, I changed the code. The current working version is under: https://gist.github.com/anonymous/fa47834db2f4f3b8b257. Any further improvements or suggestions are welcome.

What is the import error message? Could you put the backtrace here? — Mehdi Sadeghi, Aug 18 '14 at 11:50
The traceback is showing an AttributeError which seems to originate in celery-script.py, line 9. If this is the error you are referring to, please also include the content of celery-script.py. Otherwise as Mehdi Sadeghi says please incude the import error message. — Maarten, Aug 19 '14 at 8:34
celery-script.py is not part of my project but is the standard script from the celery package that is installed. — foobar, Aug 19 '14 at 8:38
Try using current app. It does not throw the exception on my machine. In tasks.py from flask import g, current_app, then use current app instead of app. remove the "from manage import app" — Maarten, Aug 19 '14 at 12:22
Thanks. You're right I can start the worker now. I started to test by sending a POST request to: 127.0.0.1:5000/test. (in the task_simple function manage.py I added a return statement). The task is sent to the worker. Worker Traceback is now above in the description. — foobar, Aug 19 '14 at 12:37

Maarten · Accepted Answer · 2014-08-21 10:36:08Z

up vote 10 down vote accepted

+50

I was off with the current_app advice.

Your celery object needs access to the application context. I found some information online about creating the Celery object with a factory function. Example below is tested without a message broker.

#factory.py
from celery import Celery
from config import config

def create_celery_app(app=None):
    app = app or create_app(config)
    celery = Celery(__name__, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask
    return celery

and in tasks.py:

#tasks.py
from factory import create_celery_app
from celery.signals import task_prerun
from flask import g

celery = create_celery_app()

@task_prerun.connect
def celery_prerun(*args, **kwargs):
    #print g
    with celery.app.app_context():
    #   # use g.db
       print g

@celery.task()
def do_some_stuff():
    with celery.app.app_context():
        # use g.db
        g.user = "test"
        print g.user

Some links:

Flask pattern for creating a Celery instance with factory function

Application using both application factory and celery

Source for said application's factory.py

Source for application tasks.py

edited Aug 21 '14 at 10:36

answered Aug 20 '14 at 14:57

Maarten

237410

Hi Maarten, thanks for your answer. With your hint I got it working with a few minor adaptions. So I would like to accept it as correct answer but nonetheless post the full working answer. Do you know what is the usual approach? – foobar Aug 21 '14 at 7:15

Happy to hear you were able to fix your code. If you want to accept the answer you can post your working code in an edit in the original question. – Maarten Aug 21 '14 at 7:36

Playing around a bit with my current solution and with your suggestion, I'm struggeling to understand why decorator needs to set explicitly the base. e.g. @celery.task(base=celery.Task) def do_some_stuff(): print g whereas just @celery.task does give an outside application context. – foobar Aug 21 '14 at 8:09

Did you use "celery.Task = ContextTask" in your factory function? Or maybe you put your working code somewhere and link to it? – Maarten Aug 21 '14 at 8:16

2

The problem with that approach is that you then create the application context directly. Which means the whole ContextTask class is kind of useless. – foobar Aug 21 '14 at 17:40

| show 9 more comments

asked	1 year ago
viewed	3742 times
active	1 year ago

current community

your communities

more stack exchange communities

Flask with create_app, SQLAlchemy and Celery

1 Answer 1

Your Answer

Not the answer you're looking for? Browse other questions tagged flask celery flask-sqlalchemy or ask your own question.

Linked

Hot Network Questions

current community

your communities

more stack exchange communities

Flask with create_app, SQLAlchemy and Celery

1 Answer 1

Your Answer

Sign up or log in

Post as a guest

Not the answer you're looking for? Browse other questions tagged flask celery flask-sqlalchemy or ask your own question.

Linked

Related

Hot Network Questions