Skip to main content

Celery support for Flask without breaking PyCharm inspections.

Project description

Even though the Flask documentation says Celery extensions are unnecessary now, I found that I still need an extension to properly use Celery in large Flask applications. Specifically I need an init_app() method to initialize Celery after I instantiate it.

This extension also comes with a single_instance method using Redis locks.

Currently only works with Redis backends.

Build Status Coverage Status Latest Version Downloads Download format License


Single instance decorator inspired by Ryan Roemer.

Supported Platforms

  • OSX and Linux.
  • Python 2.6, 2.7, 3.3, 3.4
  • Flask 0.10.1
  • Redis 2.9.1
  • Celery 3.1.11



pip install Flask-Celery-Helper


from flask import Flask
from flask.ext.celery import Celery

app = Flask('example')
app.config['REDIS_URL'] = 'redis://localhost'
celery = Celery(app)

def add_together(a, b):
    return a + b

if __name__ == '__main__':
    result = add_together.delay(23, 42)

Run these two commands in separate terminals:

celery -A example.celery worker

Factory Example

from flask.ext.celery import Celery

celery = Celery()
from flask import Flask
from extensions import celery

def create_app():
    app = Flask(__name__)
    app.config['CELERY_IMPORTS'] = ('tasks.add_together', )
    app.config['REDIS_URL'] = 'redis://localhost'
    return app
from extensions import celery

def add_together(a, b):
    return a + b
from application import create_app

app = create_app()

Single Instance Example

import time
from flask import Flask
from flask.ext.celery import Celery, single_instance
from flask.ext.redis import Redis

app = Flask('example')
app.config['REDIS_URL'] = 'redis://localhost'
celery = Celery(app)

def sleep_one_second(a, b):
    return a + b

if __name__ == '__main__':
    task1 = sleep_one_second.delay(23, 42)
    task2 = sleep_one_second.delay(20, 40)
    results1 = task1.get(propagate=False)
    results2 = task2.get(propagate=False)
    print(results1)  # 65
    if isinstance(results2, Exception) and str(results2) == 'Failed to acquire lock.':
        print('Another instance is already running.')
        print(results2)  # Should not happen.



  • Support for non-Redis backends.


  • Added Python 2.6 and 3.x support.


  • Fixed single_instance arguments with functools.


  • Added include_args argument to single_instance.


  • Initial release.

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for Flask-Celery-Helper, version 1.0.0
Filename, size File type Python version Upload date Hashes
Filename, size Flask-Celery-Helper-1.0.0.tar.gz (6.4 kB) File type Source Python version None Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page