Solution for http://stackoverflow.com/questions/25360136/flask-with-create-app-sqlalchemy-and-celery

__init__.py

# Empty

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}

SELECTED_CONFIG = "local"

extensions.py

from flask.ext.sqlalchemy import SQLAlchemy

db = SQLAlchemy()

factory.py

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

def create_celery_app(app=None):
    app = app or create_app()
    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
    celery.app = app
    return celery


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


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

    db.init_app(app)
    #celery = create_celery_app(app)

    # Register the blueprints here

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

manage.py

from factory import create_app
import tasks
app = create_app()

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

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

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)

tasks.py

from factory import create_celery_app
from celery.signals import task_prerun
from flask import g
from extensions import db

celery = create_celery_app()

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

#@celery.task(base=celery.Task)
@celery.task()
def do_some_stuff():
    print db
    print g

# run with: celery.exe worker -A tasks

This didn't work for me.

I ended up using a global class to share global objects between tasks at their startup (e.g. services connections, in my case a graphdb)