The AQL queries used in the ArangoDbBackend's implementation are potentially
vulnerable to injections because no sanity checks are performed on the
arguments used to build the query strings.

This is particularly evident when using a database collection with a dash in
its name, e.g. "celery-task-results". The query string generated by the set
method is 'INSERT {task: v}, _key: "k"} INTO celery-task-results' instead of
'INSERT {task: v}, _key: "k"} INTO `celery-task-results`' (backticks
surrounding collection name). The former is evaluated as a substraction
(celery - task - results) and is therefore an illegal collection name,
while the latter is evaluated as a string.

This commit re-implements the setter and getters using bind parameters,
which performs the necessary safety checks. Furthermore, the new query used
in the set method accounts for updates to existing keys, resolving https://github.com/celery/celery/issues/7039.

Implement safer ArangoDB queries. The AQL queries used in the ArangoDbBackend's implementation are potentially
vulnerable to injections because no sanity checks are performed on the
arguments used to build the query strings.

This is particularly evident when using a database collection with a dash in
its name, e.g. "celery-task-results". The query string generated by the set
method is 'INSERT {task: v}, _key: "k"} INTO celery-task-results' instead of
'INSERT {task: v}, _key: "k"} INTO `celery-task-results`' (backticks
surrounding collection name). The former is evaluated as a substraction
(celery - task - results) and is therefore an illegal collection name,
while the latter is evaluated as a string.

This commit re-implements the setter and getters using bind parameters,
which performs the necessary safety checks. Furthermore, the new query used
in the set method accounts for updates to existing keys, resolving https://github.com/celery/celery/issues/7039.

.. _cookbook-tasks:

================
 Task Cookbook
================

.. contents::
    :local:

.. _cookbook-task-serial:

Ensuring a task is only executed one at a time
==============================================

You can accomplish this by using a lock.

In this example we'll be using the cache framework to set a lock that's
accessible for all workers.

It's part of an imaginary RSS feed importer called `djangofeeds`.
The task takes a feed URL as a single argument, and imports that feed into
a Django model called `Feed`. We ensure that it's not possible for two or
more workers to import the same feed at the same time by setting a cache key
consisting of the MD5 check-sum of the feed URL.

The cache key expires after some time in case something unexpected happens,
and something always will...

For this reason your tasks run-time shouldn't exceed the timeout.


.. note::

    In order for this to work correctly you need to be using a cache
    backend where the ``.add`` operation is atomic.  ``memcached`` is known
    to work well for this purpose.

.. code-block:: python

    import time
    from celery import task
    from celery.utils.log import get_task_logger
    from contextlib import contextmanager
    from django.core.cache import cache
    from hashlib import md5
    from djangofeeds.models import Feed

    logger = get_task_logger(__name__)

    LOCK_EXPIRE = 60 * 10  # Lock expires in 10 minutes

    @contextmanager
    def memcache_lock(lock_id, oid):
        timeout_at = time.monotonic() + LOCK_EXPIRE - 3
        # cache.add fails if the key already exists
        status = cache.add(lock_id, oid, LOCK_EXPIRE)
        try:
            yield status
        finally:
            # memcache delete is very slow, but we have to use it to take
            # advantage of using add() for atomic locking
            if time.monotonic() < timeout_at and status:
                # don't release the lock if we exceeded the timeout
                # to lessen the chance of releasing an expired lock
                # owned by someone else
                # also don't release the lock if we didn't acquire it
                cache.delete(lock_id)

    @task(bind=True)
    def import_feed(self, feed_url):
        # The cache key consists of the task name and the MD5 digest
        # of the feed URL.
        feed_url_hexdigest = md5(feed_url).hexdigest()
        lock_id = '{0}-lock-{1}'.format(self.name, feed_url_hexdigest)
        logger.debug('Importing feed: %s', feed_url)
        with memcache_lock(lock_id, self.app.oid) as acquired:
            if acquired:
                return Feed.objects.import_feed(feed_url).url
        logger.debug(
            'Feed %s is already being imported by another worker', feed_url)

.. _cookbook-tasks:

================
 Task Cookbook
================

.. contents::
    :local:

.. _cookbook-task-serial:

Ensuring a task is only executed one at a time
==============================================

You can accomplish this by using a lock.

In this example we'll be using the cache framework to set a lock that's
accessible for all workers.

It's part of an imaginary RSS feed importer called `djangofeeds`.
The task takes a feed URL as a single argument, and imports that feed into
a Django model called `Feed`. We ensure that it's not possible for two or
more workers to import the same feed at the same time by setting a cache key
consisting of the MD5 check-sum of the feed URL.

The cache key expires after some time in case something unexpected happens,
and something always will...

For this reason your tasks run-time shouldn't exceed the timeout.


.. note::

    In order for this to work correctly you need to be using a cache
    backend where the ``.add`` operation is atomic.  ``memcached`` is known
    to work well for this purpose.

.. code-block:: python

    import time
    from celery import task
    from celery.utils.log import get_task_logger
    from contextlib import contextmanager
    from django.core.cache import cache
    from hashlib import md5
    from djangofeeds.models import Feed

    logger = get_task_logger(__name__)

    LOCK_EXPIRE = 60 * 10  # Lock expires in 10 minutes

    @contextmanager
    def memcache_lock(lock_id, oid):
        timeout_at = time.monotonic() + LOCK_EXPIRE - 3
        # cache.add fails if the key already exists
        status = cache.add(lock_id, oid, LOCK_EXPIRE)
        try:
            yield status
        finally:
            # memcache delete is very slow, but we have to use it to take
            # advantage of using add() for atomic locking
            if time.monotonic() < timeout_at and status:
                # don't release the lock if we exceeded the timeout
                # to lessen the chance of releasing an expired lock
                # owned by someone else
                # also don't release the lock if we didn't acquire it
                cache.delete(lock_id)

    @task(bind=True)
    def import_feed(self, feed_url):
        # The cache key consists of the task name and the MD5 digest
        # of the feed URL.
        feed_url_hexdigest = md5(feed_url).hexdigest()
        lock_id = '{0}-lock-{1}'.format(self.name, feed_url_hexdigest)
        logger.debug('Importing feed: %s', feed_url)
        with memcache_lock(lock_id, self.app.oid) as acquired:
            if acquired:
                return Feed.objects.import_feed(feed_url).url
        logger.debug(
            'Feed %s is already being imported by another worker', feed_url)
