Setting up large Flask application using factory pattern is very convinient, because it prevents a code being run at import time and provides more flexible way to setup application.

Celery is a good and must have tool for running asynchonious tasks, but it might be a little tricky to configure it in a large application.


Imagine a following project layout             # Async celery tasks, imports factories.celery       # Views, imports tasks
factories/   # Applicatoin factory, imports controllers        # Celery factory, imports application            # Module for worker process

Simple flask app is configured in factories.application:

from flask import Flask

from controllers import home
from .configuration import get_config

def create_application():
    config = get_config()
    app = Flask(__name__)
    return app 

In factories.celery Please note, that we need to override TaskBase, co celery can access flask app contenxt, such as current_app. :

from celery import Celery
from import Task as CeleryTask
from flask import Flask

def configure_celery(app: Flask) -> Celery:
    celery = Celery(app.import_name,
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            # Celery task will run inside app context
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask
    return celery just contains basic task. Please note, that task binded to celery instance. @shared_task could’ve been used, but this approach is less magical:

from factories.celery import create_celery
from factories.application import create_application
celery = create_celery(create_application())

def simple_task(argument):

If controllers will import there’s a circular import.


  1. Import tasks inside a view function. Violates PEP8
  2. Import tasks using importlib.
  3. Change factories.celery and tasks to point to Celery instance placeholder, which will be configured by factory.

Personally, I prefer a last option because it is clean and flexible and comes at little cost.

Introducing Celery placeholder

Let’s add file, called extensions where celery instance will be intialized at import time:

from celery import Celery

celery = Celery('celery_example', include=['myapp.tasks'])

This instance is a placeholder, it doesn’t know which broker to use, but it ill be configured in factories:

from celery import Celery
from import Task as CeleryTask
from flask import Flask

from myapp import extensions

def configure_celery(app: Flask) -> Celery:
    TaskBase: CeleryTask = extensions.celery.Task
    # Initialization of instance is not here anymore
    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    # Configuration of placeholder happens here
        # Rest of configuration
    extensions.celery.Task = ContextTask
    return extensions.celery

from myapp.extensions import celery

def simple_task(argument):

And in controllers:

from flask import Blueprint, render_template, Response, request
from myapp import tasks               # Actual tasks
from myapp.extensions import celery   # For fetching results
home = Blueprint('home_views', __name__)

@home.route('/', methods=['GET', 'POST'])
def index() -> Response:
    task_id = None
    if request.method == 'POST':
        task_id = tasks.simple_task.delay(request.form.get('message'))
    return render_template('index.html', task_id=task_id)

def task_result(task_id):
    result = celery.AsyncResult(task_id)
    return render_template('task_result.html', task_id=task_id, result=result)

And file with celery worker:

from myapp.factories.application import create_application
from myapp.factories.celery import configure_celery

celery = configure_celery(create_application())


The only disadvantage of this approach is that celery instance is singleton, which is run at import time, but it should not be a problem for the most of the applications.

Working proof of concept is available on github