Skip to content

Instantly share code, notes, and snippets.

@eduzen
Created October 13, 2019 12:50
Show Gist options
  • Save eduzen/12e59795abdfaaa021e4d14a3f94c03d to your computer and use it in GitHub Desktop.
Save eduzen/12e59795abdfaaa021e4d14a3f94c03d to your computer and use it in GitHub Desktop.
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment