single_instance_task.py

import redis

from django.core.cache import cache
from functools import wraps

def single_instance_task(function=None, key="", timeout=None):
    """Enforce only one celery task at a time."""
    
    @wraps(func)
    def _task_execution(run_func):
        """Decorator."""

        def _caller(*args, **kwargs):
            """Caller."""
            ret_value = None
            with cache.lock(key, timeout=timeout) as lock:
                ret_value = run_func(*args, **kwargs)
            return ret_value

        return _caller

    return _task_execution(function) if function is not None else _task_execution