celery_task_with_redis_lock.py

redis_connection = StrictRedis(host='redis')

# source: https://redis.io/topics/distlock
# This will be used to release lock IF AND ONLY IF we have acquired it
REMOVE_ONLY_IF_OWNER_SCRIPT = \
"""if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end
"""

@contextmanager
def redis_lock(lock_name, expires=60):
    # first, get a random value (for ownership)
    random_value = str(uuid.uuid4())
    # execute SET {lock_name} {random_value} EX {expires} NX
    lock_acquired = bool(
        redis_connection.set(lock_name, random_value, ex=expires, nx=True)
    )
    logger.info(f'Lock acquired? {lock_name} for {expires} - {lock_acquired}')

    yield lock_acquired

    if lock_acquired:
        # if lock was acquired, then try to release it BUT ONLY if we are the owner
        # (i.e. value inside is identical to what we put there originally)
        redis_connection.eval(REMOVE_ONLY_IF_OWNER_SCRIPT, 1, lock_name, random_value)
        logger.info(f'Lock {lock_name} released!')


@app.task()
def periodic_task_with_lock():
    """
    This task uses redis_lock context manager to ensure it is executed
    only one at a time.
    """
    with redis_lock('periodic_task_with_lock_lock', 10) as acquired:
        if not acquired:
            # abort if one failed to acquire the lock
            logger.warning('Lock not acquired!')
            return

        # ok, lock acquired! Work as usual...
        will_sleep_for = randint(5, 8)
        logger.info(f'I will keep lock for {will_sleep_for}s')
        sleep(will_sleep_for)