Skip to content

Instantly share code, notes, and snippets.

/__init__.py

Created Aug 21, 2014
Embed
What would you like to do?
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"
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'])
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
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()
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()
@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
@pdonorio

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.