from app import celery
__all__ = ["execute_print_job"]
@celery.task()
def execute_print_job():
print("Execute print job.")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()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# 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 *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 appfrom 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", 201from 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()<!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>flask
flask-restful
celery
redis
python -m venv venv
source venv/bin/activate
Assuming Redis is installed and running on localhost using port 6379.
python run.py
celery -A app.celery_worker.celery worker --loglevel=info
Based on and inspired by Stefano Frassetto's article Flask + Celery = How To.