This is a pre-production deployment of Warehouse, however changes made here WILL affect the production instance of PyPI.
Latest Version Dependencies status unknown Test status unknown Test coverage unknown
Project Description


The Celery daemon needs to be restarted every time an existing task is modified or new tasks are added.


Use only one Celery task that’s generic enough to run arbitrary Python functions with arbitrary arguments and wrap this task in a custom decorator.

As a bonus, the job will still work (synchronously) when Celery/the broker are not running.

The job’s calling API is the same as Celery’s: .s(), .delay() and .apply_async()


  • in which is in the same directory as, create a Celery app and then import the custom decorator:
import celery
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '.'))
#os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings')
app = celery.Celery()

from generic_celery_task.decorators import task
  • start Celery along with its broker. You no longer need to restart Celery after this.
  • in another file, importable from, create your task:
from celeryapp import task

def job(x, y):
    return 'x + y = %d' % (x + y)
  • now use it:
import celery

# direct function call
assert job(1, 2) == 'x + y = 3'

# using the delay() method
res = job.delay(1, 2)
# if the Celery daemon and its backend broker are running, 'res' is an instance of AsyncResult
if isinstance(res, celery.result.AsyncResult):
    assert res.get() == 'x + y = 3'
    # if either one is not running, a warning is issued and the function is executed synchronously
    # with 'res' being the actual function result
    # if you want to silence the warnings, use '@task(quiet=True)'
    assert res == 'x + y = 3'

# using the apply_async() method
res = job.apply_async(args=[1, 2])
# and process the result as above, if you need to


A is provided. You know what to do with it.


The tests require nose, redis, redis-py and assume that the port 6389 is free.

Run the tests with “python test” or with “nosetests -v”.

This package was tested with python-2.7.6, python-3.3.4, nose-1.3.0, celery-3.1.10, redis-2.8.7 and redis-py-2.9.1 .


  • the module which holds your custom task will be reloaded. If it contains a class using ‘super’ and its instance, you might run into the problem described here. Apply one of the proposed fixes.
  • the state of the Celery daemon and its broker are checked only once, when the first .delay() or .apply_async() method is called on a custom task.
Release History

Release History


This version

History Node

TODO: Figure out how to actually get changelog content.

Changelog content for this version goes here.

Donec et mollis dolor. Praesent et diam eget libero egestas mattis sit amet vitae augue. Nam tincidunt congue enim, ut porta lorem lacinia consectetur. Donec ut libero sed arcu vehicula ultricies a non tortor. Lorem ipsum dolor sit amet, consectetur adipiscing elit.

Show More

Download Files

Download Files

TODO: Brief introduction on what you do with files - including link to relevant help section.

File Name & Checksum SHA256 Checksum Help Version File Type Upload Date
generic_celery_task-0.2.tar.gz (5.6 kB) Copy SHA256 Checksum SHA256 Source Apr 2, 2014

Supported By

WebFaction WebFaction Technical Writing Elastic Elastic Search Pingdom Pingdom Monitoring Dyn Dyn DNS HPE HPE Development Sentry Sentry Error Logging CloudAMQP CloudAMQP RabbitMQ Heroku Heroku PaaS Kabu Creative Kabu Creative UX & Design Fastly Fastly CDN DigiCert DigiCert EV Certificate Rackspace Rackspace Cloud Servers DreamHost DreamHost Log Hosting