Skip to content

Instantly share code, notes, and snippets.

@thepacketgeek
Forked from anonymous/__init__.py
Last active August 29, 2015 14:18
Show Gist options
  • Save thepacketgeek/d1d72275ce5404233495 to your computer and use it in GitHub Desktop.
Save thepacketgeek/d1d72275ce5404233495 to your computer and use it in GitHub Desktop.
from extensions import db
from celery import Celery
from flask import Flask
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_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
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 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 extensions import db
celery = create_celery_app()
@task_prerun.connect
def celery_prerun(*args, **kwargs):
with celery.app.app_context():
# # use g.db
print db
#@celery.task(base=celery.Task)
@celery.task()
def do_some_stuff():
print db
# run with: celery.exe worker -A tasks
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment