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()
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() app.config_from_object('celeryconfig') from generic_celery_task.decorators import task
from celeryapp import task @task def job(x, y): return 'x + y = %d' % (x + y)
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' else: # 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 setup.py 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 setup.py 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 .
TODO: Figure out how to actually get changelog content.
Changelog content for this version goes here.