Using Celery with Flask

In this post, I will talk about using Celery with Flask. I’m going to assume that you have completed this tutorial on Celery.

Basic Flask app structure

Assume that we have written a nicely working Flask app which now needs to do some backend processing.

Here’s our basic Flask app structure:

+--myapp_dir/
   +--app/
      +--static/
      +--templates/
      +--utils/
      +--views/
      +--__init__.py
   +--config.py
   +--run.py
   +--requirements.txt

Include Celery

Including Celery into the current mix, it’d look something like –

+--myapp_dir/
   +--app/
      +--static/
      +--templates/
      +--utils/
      +--views/
      +--app_tasks.py
      +--__init__.py
   +--config.py
   +--run.py
   +--requirements.txt
   +--celery_tasks.py

Notice that we have now added two new files – celery_tasks.py and app_tasks.py

app_tasks.py

In this file would reside our Celery tasks. Let’s put some sample code –

from celery import Celery
import os
os.environ['CELERY_CONFIG_MODULE'] = 'celeryconfig'

app = Celery()

@app.task
def add(x, y):
    return x + y

@app.task
def print_results(res=None):
    if res:
        print "Results: %s" % res
    else:
        print "Bello Merld"

Here are some really catchy facts about this code:

• os.environ['CELERY_CONFIG_MODULE'] = 'celeryconfig' – This line expects you to have a file called celeryconfig in the same directory as your app_tasks.py. There are 3 ways of handling configuration in Celery and it is documented here.
• return x + y – This line looks like a very simple function return but when you don’t have a result backend, it throws – NotImplementedError

So, is something wrong with the code?

The code is not wrong but there are 2 advices you might want to bear in mind. These are more like my best practises.

Advice 1 – Do not create a separate settings file when using Celery with Flask

If you look back at the directory structure you will find that we have a file config.py. Re-use your config.py and do not store multiple settings in multiple files.

Example usage:

app = Celery()
app.config_from_object('config.ProductionConfig')

I tend to create my config.py in the following way:

class Config(object):
    # All defaults come here.
    MYSQL_HOST = 1.2.3.4
    MYSQL_USER = 'user'
    MYSQL_PASSWORD = 'password'

class ProductionConfig(Config):
    # All production specific settings can be overriden here.
    MYSQL_HOST = 127.0.0.1
    MYSQL_USER = 'master_user'
    MYSQL_PASSWORD = 'master_password'

class DevelopmentConfig(Config):
    # All development specific settings can be overriden here.
    MYSQL_HOST = 192.168.1.3

This structure allows me to quickly switch between my Production and Development configurations.

Here’s how I can add new settings directly or update existing settings –

app = Celery()
app.config_from_object('config.ProductionConfig')
app.conf.update(CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/')

Advice 2 – Include a Celery results backend before you return anything

Set the following variable – CELERY_RESULT_BACKEND. The error that you see when you don’t set it is –

NotImplementedError: No result backend configured. Please see the documentation for more information.

Note that implementing a results backend could be dangerous as described in my other post.

Updated app_tasks.py

Based on the above advices, let us modify the app_tasks.py.

from celery import Celery

app = Celery()
app.config_from_object('config.ProductionConfig')
app.conf.update(CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/')

# Ensure that the task expires quickly so that it doesn't slow Redis in case of
# high usage.
app.conf.update(CELERY_TASK_RESULT_EXPIRES=600)

@app.task
def add(x, y):
    return x + y

@app.task
def print_results(res=None):
    if res:
        print "Results: %s" % res
    else:
        print "Bello Merld"

celery_tasks.py

In this file, we would import all the tasks from app_tasks. This helps us maintain Celery tasks per app.

from app.app_tasks import *

Call Celery Tasks in Flask

To now use these elsewhere in Flask, we could –

from app.app_tasks import print_results

@app.route('/fetch_results/')
def fetch_results():
    # do something
    print_results.delay(res)
    return jsonify(message=res)

In a later post, I will cover how to deploy Celery with Flask.