Code files for http://mpcabd.igeex.biz/python-celery-asynchronous-task-decorator/

celery_decorators.py

# This work is licensed under the GNU Public License (GPL).
# To view a copy of this license, visit http://www.gnu.org/copyleft/gpl.html

# For more information visit this blog post http://mpcabd.igeex.biz/python-celery-asynchronous-task-decorator/

# Written by Abd Allah Diab (mpcabd)
# Email: mpcabd ^at^ gmail ^dot^ com
# Website: http://mpcabd.igeex.biz

from django.utils.decorators import available_attrs
from django.utils.functional import wraps
 
def asynchronous_task(check_if_active = False):
    def wrapper(f):
        try:
            @wraps(f, assigned=available_attrs(f))
            def _wrapper(*args, **kwargs):
                try:
                    if check_if_active:
                        from celery.task.control import inspect
                        insp = inspect()
                        d = insp.active()
                        if not d:
                            raise Exception
                    from celery_tasks import execute
                    execute.delay(f.__module__, f.func_name, *args, **kwargs)
                except Exception as e:
                    f(*args, **kwargs)
            _wrapper.func = f
            return _wrapper
        except Exception:
            return f
    return wrapper

celery_tasks.py

# This work is licensed under the GNU Public License (GPL).
# To view a copy of this license, visit http://www.gnu.org/copyleft/gpl.html

# For more information visit this blog post http://mpcabd.igeex.biz/python-celery-asynchronous-task-decorator/

# Written by Abd Allah Diab (mpcabd)
# Email: mpcabd ^at^ gmail ^dot^ com
# Website: http://mpcabd.igeex.biz

import os, sys
from celery.task import task
 
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
sys.path.insert(0, os.path.dirname(__file__))

##########################
# Uncomment the following code block to enable it to work with Django.
# Either put this file next to the settings.py file, or change the block code to import your settings module from the correct path.
##########################
 
#from django.core.management import setup_environ
#import settings
#setup_environ(settings)
 
@task(name='execute')
def execute(module_name, method_name, *args, **kwargs):
    try:
        m = __import__(module_name)
        if m.__name__ != module_name:
            modules = module_name.split('.')[1:]
            for module in modules:
                m = getattr(m, module)
        f = getattr(m, method_name)
        if hasattr(f, 'func'):
            f = f.func
        return f(*args, **kwargs)
    except Exception, e:
        raise e