Skip to content

Instantly share code, notes, and snippets.


Created Aug 21, 2014
What would you like to do?
import os
basedir = os.path.abspath(os.path.dirname(__file__))
class Config:
def init_app(app):
class LocalConfig(Config):
DEBUG = True
SQLALCHEMY_DATABASE_URI = r"sqlite:///" + os.path.join(basedir,
CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'
config = {
"local": LocalConfig}
from flask.ext.sqlalchemy import SQLAlchemy
db = SQLAlchemy()
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'])
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 = app
return celery
def create_before_request(app):
def before_request():
g.db = db
return before_request
def create_app():
app = Flask(__name__)
#celery = create_celery_app(app)
# Register the blueprints here
# Add the before request handler
return app
from factory import create_app
import tasks
app = create_app()
@app.route('/test', methods=['POST'])
def task_simple():
return ""
if __name__ == "__main__":
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)
from factory import create_celery_app
from celery.signals import task_prerun
from flask import g
from extensions import db
celery = create_celery_app()
def celery_prerun(*args, **kwargs):
#print g
# # use g.db
print db
print g
def do_some_stuff():
print db
print g
# run with: celery.exe worker -A tasks

This comment has been minimized.

Copy link

@pdonorio pdonorio commented Jul 8, 2016

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)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.