moosetraveller/flask-celery.md

## flask-celery.md

      
    Raw
  

              flask-celery.md
            
          
    Flask with Celery

Files

tasks/long_running.py

from app import celery

__all__ = ["execute_print_job"]

@celery.task()
def execute_print_job():
    print("Execute print job.")
app/__init__.py

import os
from celery import Celery

def make_celery(app_name: str=__name__) -> Celery:
    redis_uri = os.getenv("REDIS_URI", "redis://localhost:6379")
    return Celery(app_name, backend=redis_uri, broker=redis_uri)

celery = make_celery()
app/celery_utils.py

from celery import Celery
from flask import Flask

def init_celery(celery: Celery, app: Flask) -> None:

    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
app/celery_worker.py

# start with 
# celery -A app.celery_worker.celery worker --loglevel=debug

from app import celery
from app.factory import create_app
from app.celery_utils import init_celery

app = create_app()
init_celery(celery, app)

from tasks.long_running import *
app/factory.py

from distutils.command.config import config
from flask import Flask
from celery import Celery

from .celery_utils import init_celery

from pathlib import Path

PKG_NAME = Path(__file__).parent.name

def create_app(secret_key: str, app_name: str=PKG_NAME, celery: Celery=None) -> Flask:

    # template folder is relative to module, however, we use ../templates
    app = Flask(app_name, template_folder="../templates")
    
    app.config["SECRET_KEY"] = secret_key
    
    if celery:
        init_celery(celery, app)

    return app
resources/print_jobs.py

from flask_restful import Resource

from tasks.long_running import execute_print_job

class PrintJobs(Resource):

    @classmethod
    def post(cls):
        execute_print_job.delay()
        return "Print Job created and started", 201
run.py

from flask import render_template
from flask_restful import Api

from app import factory
import app as flask_app

from resources.print_jobs import PrintJobs

# python -c "import secrets; print(secrets.token_hex(16))"
SECRET_KEY = "3c27425f792cb3f5651368b2c44095a9"

def run():
    
    app = factory.create_app(SECRET_KEY, celery=flask_app.celery)
    api = Api(app)
    
    api.add_resource(PrintJobs, "/print_jobs")
    
    @app.route("/")
    def index():
        return render_template("index.html")
        
    app.run(port=5000)
    
    
if __name__ == "__main__":
    run()
templates/index.hml

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Hello, World!</title>
</head>
<body>
    Hello, World!
</body>
</html>
requirements.txt

flask
flask-restful
celery
redis

Environment

python -m venv venv
source venv/bin/activate

Assuming Redis is installed and running on localhost using port 6379.
Run

Run Application

python run.py

Run Celery Worker

celery -A app.celery_worker.celery worker --loglevel=info

Credits

Based on and inspired by Stefano Frassetto's article Flask + Celery = How To.