added arangodb backend (celery#5217)

* added arangodb backend

* changed to Arangodb

* removed unnecessary

* minor fix for delete

* Added basic documentation

* added trailing comma

* replaced . with :

* minor fixes in mget

* order for url in __init__

* added documentation

* updated minimum pyArango requirements and minor log changes

* added http protocol backend settings

* fixed arangodb sphinx error

* added testcases for arangodb

* fixed flake8 errors

* db as just property

* key as str

README.rst

@@ -313,6 +313,9 @@ Transports and Backends
 :``celery[couchbase]``:
     for using Couchbase as a result backend.
 
+:``celery[arangodb]``:
+    for using ArangoDB as a result backend.
+
 :``celery[elasticsearch]``:
     for using Elasticsearch as a result backend.
 

celery/app/backends.py

@@ -37,6 +37,7 @@
     'consul': 'celery.backends.consul:ConsulBackend',
     'dynamodb': 'celery.backends.dynamodb:DynamoDBBackend',
     'azureblockblob': 'celery.backends.azureblockblob:AzureBlockBlobBackend',
+    'arangodb': 'celery.backends.arangodb:ArangoDbBackend',
     's3': 'celery.backends.s3:S3Backend',
 }
 

celery/app/defaults.py

@@ -158,6 +158,10 @@ def __repr__(self):
 
         backend_settings=Option(None, type='dict'),
     ),
+    arangodb=Namespace(
+        __old__=old_ns('celery_arangodb'),
+        backend_settings=Option(None, type='dict')
+    ),
     mongodb=Namespace(
         __old__=old_ns('celery_mongodb'),
 

celery/backends/arangodb.py

@@ -0,0 +1,211 @@
+# -*- coding: utf-8 -*-
+"""ArangoDb result store backend."""
+
+# pylint: disable=W1202,W0703
+
+from __future__ import absolute_import, unicode_literals
+
+import logging
+import json
+from kombu.utils.url import _parse_url
+from kombu.utils.encoding import str_t
+
+from celery.exceptions import ImproperlyConfigured
+
+from .base import KeyValueStoreBackend
+
+try:
+    from pyArango import connection as py_arango_connection
+    from pyArango.theExceptions import AQLQueryError
+except ImportError:
+    py_arango_connection = AQLQueryError = None   # noqa
+
+__all__ = ('ArangoDbBackend',)
+
+
+class ArangoDbBackend(KeyValueStoreBackend):
+    """ArangoDb backend.
+
+    Sample url
+    "arangodb://username:password@host:port/database/collection"
+    *arangodb_backend_settings* is where the settings are present
+    (in the app.conf)
+    Settings should contain the host, port, username, password, database name,
+    collection name else the default will be chosen.
+    Default database name and collection name is celery.
+
+    Raises
+    ------
+    celery.exceptions.ImproperlyConfigured:
+        if module :pypi:`pyArango` is not available.
+
+    """
+
+    host = '127.0.0.1'
+    port = '8529'
+    database = 'celery'
+    collection = 'celery'
+    username = None
+    password = None
+    # protocol is not supported in backend url (http is taken as default)
+    http_protocol = 'http'
+
+    # Use str as arangodb key not bytes
+    key_t = str_t
+
+    def __init__(self, url=None, *args, **kwargs):
+        """Parse the url or load the settings from settings object."""
+        super(ArangoDbBackend, self).__init__(*args, **kwargs)
+
+        if py_arango_connection is None:
+            raise ImproperlyConfigured(
+                'You need to install the pyArango library to use the '
+                'ArangoDb backend.',
+            )
+
+        self.url = url
+
+        if url is None:
+            host = port = database = collection = username = password = None
+        else:
+            (
+                _schema, host, port, username, password,
+                database_collection, _query
+            ) = _parse_url(url)
+            if database_collection is None:
+                database = collection = None
+            else:
+                database, collection = database_collection.split('/')
+
+        config = self.app.conf.get('arangodb_backend_settings', None)
+        if config is not None:
+            if not isinstance(config, dict):
+                raise ImproperlyConfigured(
+                    'ArangoDb backend settings should be grouped in a dict',
+                )
+        else:
+            config = {}
+
+        self.host = host or config.get('host', self.host)
+        self.port = int(port or config.get('port', self.port))
+        self.http_protocol = config.get('http_protocol', self.http_protocol)
+        self.database = database or config.get('database', self.database)
+        self.collection = \
+            collection or config.get('collection', self.collection)
+        self.username = username or config.get('username', self.username)
+        self.password = password or config.get('password', self.password)
+        self.arangodb_url = "{http_protocol}://{host}:{port}".format(
+            http_protocol=self.http_protocol, host=self.host, port=self.port
+        )
+        self._connection = None
+
+    @property
+    def connection(self):
+        """Connect to the arangodb server."""
+        if self._connection is None:
+            self._connection = py_arango_connection.Connection(
+                arangoURL=self.arangodb_url, username=self.username,
+                password=self.password
+            )
+        return self._connection
+
+    @property
+    def db(self):
+        """Database Object to the given database."""
+        return self.connection[self.database]
+
+    def get(self, key):
+        try:
+            logging.debug(
+                'RETURN DOCUMENT("{collection}/{key}").task'.format(
+                    collection=self.collection, key=key
+                )
+            )
+            query = self.db.AQLQuery(
+                'RETURN DOCUMENT("{collection}/{key}").task'.format(
+                    collection=self.collection, key=key
+                )
+            )
+            result = query.response["result"][0]
+            if result is None:
+                return None
+            return json.dumps(result)
+        except AQLQueryError as aql_err:
+            logging.error(aql_err)
+            return None
+        except Exception as err:
+            logging.error(err)
+            return None
+
+    def set(self, key, value):
+        """Insert a doc with value into task attribute and _key as key."""
+        try:
+            logging.debug(
+                'INSERT {{ task: {task}, _key: "{key}" }} INTO {collection}'
+                .format(
+                    collection=self.collection, key=key, task=value
+                )
+            )
+            self.db.AQLQuery(
+                'INSERT {{ task: {task}, _key: "{key}" }} INTO {collection}'
+                .format(
+                    collection=self.collection, key=key, task=value
+                )
+            )
+        except AQLQueryError as aql_err:
+            logging.error(aql_err)
+        except Exception as err:
+            logging.error(err)
+
+    def mget(self, keys):
+        try:
+            json_keys = json.dumps(keys)
+            logging.debug(
+                """
+                FOR key in {keys}
+                    RETURN DOCUMENT(CONCAT("{collection}/", key).task
+                """.format(
+                    collection=self.collection, keys=json_keys
+                )
+            )
+            query = self.db.AQLQuery(
+                """
+                FOR key in {keys}
+                    RETURN DOCUMENT(CONCAT("{collection}/", key).task
+                """.format(
+                    collection=self.collection, keys=json_keys
+                )
+            )
+            results = []
+            while True:
+                results.extend(query.response['result'])
+                query.nextBatch()
+        except StopIteration:
+            values = [
+                result if result is None else json.dumps(result)
+                for result in results
+            ]
+            return values
+        except AQLQueryError as aql_err:
+            logging.error(aql_err)
+            return [None] * len(keys)
+        except Exception as err:
+            logging.error(err)
+            return [None] * len(keys)
+
+    def delete(self, key):
+        try:
+            logging.debug(
+                'REMOVE {{ _key: "{key}" }} IN {collection}'.format(
+                    key=key, collection=self.collection
+                )
+            )
+            self.db.AQLQuery(
+                'REMOVE {{ _key: "{key}" }} IN {collection}'.format(
+                    key=key, collection=self.collection
+                )
+            )
+        except AQLQueryError as aql_err:
+            logging.error(aql_err)
+        except Exception as err:
+            logging.error(err)

docs/includes/installation.txt

@@ -82,6 +82,9 @@ Transports and Backends
 :``celery[couchbase]``:
     for using Couchbase as a result backend.
 
+:``celery[arangodb]``:
+    for using ArangoDB as a result backend.
+
 :``celery[elasticsearch]``:
     for using Elasticsearch as a result backend.
 

docs/internals/reference/celery.backends.arangodb.rst

@@ -0,0 +1,11 @@
+============================================
+ ``celery.backends.arangodb``
+============================================
+
+.. contents::
+    :local:
+.. currentmodule:: celery.backends.arangodb
+
+.. automodule:: celery.backends.arangodb
+    :members:
+    :undoc-members:

docs/internals/reference/index.rst

@@ -36,6 +36,7 @@
     celery.backends.riak
     celery.backends.cassandra
     celery.backends.couchbase
+    celery.backends.arangodb
     celery.backends.dynamodb
     celery.backends.filesystem
     celery.backends.cosmosdbsql

docs/spelling_wordlist.txt

@@ -18,6 +18,7 @@ Andreas
 Andrey
 Andriy
 Aneil
+ArangoDB
 Areski
 Armin
 Artyom

docs/userguide/configuration.rst

@@ -96,6 +96,7 @@ have been moved into a new  ``task_`` prefix.
 ``S3_ENDPOINT_URL``                    :setting:`s3_endpoint_url`
 ``S3_REGION``                          :setting:`s3_region`
 ``CELERY_COUCHBASE_BACKEND_SETTINGS``  :setting:`couchbase_backend_settings`
+``CELERY_ARANGODB_BACKEND_SETTINGS``   :setting:`arangodb_backend_settings`
 ``CELERY_MONGODB_BACKEND_SETTINGS``    :setting:`mongodb_backend_settings`
 ``CELERY_EVENT_QUEUE_EXPIRES``         :setting:`event_queue_expires`
 ``CELERY_EVENT_QUEUE_TTL``             :setting:`event_queue_ttl`
@@ -607,6 +608,10 @@ Can be one of the following:
     Use `Couchbase`_ to store the results.
     See :ref:`conf-couchbase-result-backend`.
 
+* ``arangodb``
+    Use `ArangoDB`_ to store the results.
+    See :ref:`conf-arangodb-result-backend`.
+
 * ``couchdb``
     Use `CouchDB`_ to store the results.
     See :ref:`conf-couchdb-result-backend`.
@@ -645,6 +650,7 @@ Can be one of the following:
 .. _`CouchDB`: http://www.couchdb.com/
 .. _`CosmosDB`: https://azure.microsoft.com/en-us/services/cosmos-db/
 .. _`Couchbase`: https://www.couchbase.com/
+.. _`ArangoDB`: https://www.arangodb.com/
 .. _`Consul`: https://consul.io/
 .. _`AzureBlockBlob`: https://azure.microsoft.com/en-us/services/storage/blobs/
 .. _`S3`: https://aws.amazon.com/s3/
@@ -1613,6 +1619,66 @@ This is a dict supporting the following keys:
 
     Password to authenticate to the Couchbase server (optional).
 
+.. _conf-arangodb-result-backend:
+
+ArangoDB backend settings
+--------------------------
+
+.. note::
+
+    The ArangoDB backend requires the :pypi:`pyArango` library.
+
+    To install this package use :command:`pip`:
+
+    .. code-block:: console
+
+        $ pip install celery[arangodb]
+
+    See :ref:`bundles` for instructions how to combine multiple extension
+    requirements.
+
+This backend can be configured via the :setting:`result_backend`
+set to a ArangoDB URL:
+
+.. code-block:: python
+
+    result_backend = 'arangodb://username:password@host:port/database/collection'
+
+.. setting:: arangodb_backend_settings
+
+``arangodb_backend_settings``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Default: ``{}`` (empty mapping).
+
+This is a dict supporting the following keys:
+
+* ``host``
+
+    Host name of the ArangoDB server. Defaults to ``localhost``.
+
+* ``port``
+
+    The port the ArangoDB server is listening to. Defaults to ``8529``.
+
+* ``database``
+
+    The default database in the ArangoDB server is writing to.
+    Defaults to ``celery``.
+
+* ``collection``
+
+    The default collection in the ArangoDB servers database is writing to.
+    Defaults to ``celery``.
+
+* ``username``
+
+    User name to authenticate to the ArangoDB server as (optional).
+
+* ``password``
+
+    Password to authenticate to the ArangoDB server (optional).
+
 .. _conf-cosmosdbsql-result-backend:
 
 CosmosDB backend settings (experimental)

requirements/extras/arangodb.txt

@@ -0,0 +1 @@
+pyArango>=1.3.2

requirements/test-ci-default.txt

@@ -14,6 +14,7 @@
 -r extras/elasticsearch.txt
 -r extras/couchdb.txt
 -r extras/couchbase.txt
+-r extras/arangodb.txt
 -r extras/consul.txt
 -r extras/cosmosdbsql.txt
 -r extras/cassandra.txt

setup.py

@@ -57,6 +57,7 @@ def _pyimp():
     'memcache',
     'pymemcache',
     'couchbase',
+    'arangodb',
     'eventlet',
     'gevent',
     'msgpack',