Flask Celery

Background Tasks with Celery

pip install celery redis

from celery import Celery

def make_celery(app):
    celery = Celery(app.import_name, broker=app.config["CELERY_BROKER_URL"])
    celery.conf.update(app.config)
    TaskBase = celery.Task

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

    celery.Task = ContextTask
    return celery

celery = make_celery(app)

# Define task
@celery.task
def send_welcome_email(user_id):
    user = db.session.get(User, user_id)
    # send email...

# Dispatch from route
@app.route("/register", methods=["POST"])
def register():
    user = create_user(request.json)
    send_welcome_email.delay(user.id)   # non-blocking!
    return jsonify(user.to_dict()), 201

# Start worker
# celery -A app.celery worker --loglevel=info