We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
MongoDB backend does not support mongodb+srv:// URL's
## Checklist

https://github.com/celery/celery/blob/master/celery/backends/mongodb.py#L143-L146

## Steps to reproduce

Set the MongoDB URL to one like:

```mongodb+srv://mongo.private.corp.example.com/celery?ssl=false```

## Expected behavior

This works.

## Actual behavior

This fails because the URL parsing does not match on `mongodb+srv://` instead trying to match `mongodb://` only.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `celery/backends/mongodb.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 """MongoDB result store backend."""
3 from __future__ import absolute_import, unicode_literals
4 
5 from datetime import datetime, timedelta
6 
7 from kombu.exceptions import EncodeError
8 from kombu.utils.objects import cached_property
9 from kombu.utils.url import maybe_sanitize_url
10 
11 from celery import states
12 from celery.exceptions import ImproperlyConfigured
13 from celery.five import items, string_t
14 
15 from .base import BaseBackend
16 
17 try:
18     import pymongo
19 except ImportError:  # pragma: no cover
20     pymongo = None   # noqa
21 
22 if pymongo:
23     try:
24         from bson.binary import Binary
25     except ImportError:                     # pragma: no cover
26         from pymongo.binary import Binary   # noqa
27     from pymongo.errors import InvalidDocument  # noqa
28 else:                                       # pragma: no cover
29     Binary = None                           # noqa
30 
31     class InvalidDocument(Exception):       # noqa
32         pass
33 
34 __all__ = ('MongoBackend',)
35 
36 BINARY_CODECS = frozenset(['pickle', 'msgpack'])
37 
38 
39 class MongoBackend(BaseBackend):
40     """MongoDB result backend.
41 
42     Raises:
43         celery.exceptions.ImproperlyConfigured:
44             if module :pypi:`pymongo` is not available.
45     """
46 
47     mongo_host = None
48     host = 'localhost'
49     port = 27017
50     user = None
51     password = None
52     database_name = 'celery'
53     taskmeta_collection = 'celery_taskmeta'
54     groupmeta_collection = 'celery_groupmeta'
55     max_pool_size = 10
56     options = None
57 
58     supports_autoexpire = False
59 
60     _connection = None
61 
62     def __init__(self, app=None, **kwargs):
63         self.options = {}
64 
65         super(MongoBackend, self).__init__(app, **kwargs)
66 
67         if not pymongo:
68             raise ImproperlyConfigured(
69                 'You need to install the pymongo library to use the '
70                 'MongoDB backend.')
71 
72         # Set option defaults
73         for key, value in items(self._prepare_client_options()):
74             self.options.setdefault(key, value)
75 
76         # update conf with mongo uri data, only if uri was given
77         if self.url:
78             if self.url == 'mongodb://':
79                 self.url += 'localhost'
80 
81             uri_data = pymongo.uri_parser.parse_uri(self.url)
82             # build the hosts list to create a mongo connection
83             hostslist = [
84                 '{0}:{1}'.format(x[0], x[1]) for x in uri_data['nodelist']
85             ]
86             self.user = uri_data['username']
87             self.password = uri_data['password']
88             self.mongo_host = hostslist
89             if uri_data['database']:
90                 # if no database is provided in the uri, use default
91                 self.database_name = uri_data['database']
92 
93             self.options.update(uri_data['options'])
94 
95         # update conf with specific settings
96         config = self.app.conf.get('mongodb_backend_settings')
97         if config is not None:
98             if not isinstance(config, dict):
99                 raise ImproperlyConfigured(
100                     'MongoDB backend settings should be grouped in a dict')
101             config = dict(config)  # don't modify original
102 
103             if 'host' in config or 'port' in config:
104                 # these should take over uri conf
105                 self.mongo_host = None
106 
107             self.host = config.pop('host', self.host)
108             self.port = config.pop('port', self.port)
109             self.mongo_host = config.pop('mongo_host', self.mongo_host)
110             self.user = config.pop('user', self.user)
111             self.password = config.pop('password', self.password)
112             self.database_name = config.pop('database', self.database_name)
113             self.taskmeta_collection = config.pop(
114                 'taskmeta_collection', self.taskmeta_collection,
115             )
116             self.groupmeta_collection = config.pop(
117                 'groupmeta_collection', self.groupmeta_collection,
118             )
119 
120             self.options.update(config.pop('options', {}))
121             self.options.update(config)
122 
123     def _prepare_client_options(self):
124         if pymongo.version_tuple >= (3,):
125             return {'maxPoolSize': self.max_pool_size}
126         else:  # pragma: no cover
127             return {'max_pool_size': self.max_pool_size,
128                     'auto_start_request': False}
129 
130     def _get_connection(self):
131         """Connect to the MongoDB server."""
132         if self._connection is None:
133             from pymongo import MongoClient
134 
135             host = self.mongo_host
136             if not host:
137                 # The first pymongo.Connection() argument (host) can be
138                 # a list of ['host:port'] elements or a mongodb connection
139                 # URI.  If this is the case, don't use self.port
140                 # but let pymongo get the port(s) from the URI instead.
141                 # This enables the use of replica sets and sharding.
142                 # See pymongo.Connection() for more info.
143                 host = self.host
144                 if isinstance(host, string_t) \
145                    and not host.startswith('mongodb://'):
146                     host = 'mongodb://{0}:{1}'.format(host, self.port)
147             # don't change self.options
148             conf = dict(self.options)
149             conf['host'] = host
150 
151             self._connection = MongoClient(**conf)
152 
153         return self._connection
154 
155     def encode(self, data):
156         if self.serializer == 'bson':
157             # mongodb handles serialization
158             return data
159         payload = super(MongoBackend, self).encode(data)
160 
161         # serializer which are in a unsupported format (pickle/binary)
162         if self.serializer in BINARY_CODECS:
163             payload = Binary(payload)
164         return payload
165 
166     def decode(self, data):
167         if self.serializer == 'bson':
168             return data
169         return super(MongoBackend, self).decode(data)
170 
171     def _store_result(self, task_id, result, state,
172                       traceback=None, request=None, **kwargs):
173         """Store return value and state of an executed task."""
174         meta = {
175             '_id': task_id,
176             'status': state,
177             'result': self.encode(result),
178             'date_done': datetime.utcnow(),
179             'traceback': self.encode(traceback),
180             'children': self.encode(
181                 self.current_task_children(request),
182             ),
183         }
184         if request and getattr(request, 'parent_id', None):
185             meta['parent_id'] = request.parent_id
186 
187         try:
188             self.collection.save(meta)
189         except InvalidDocument as exc:
190             raise EncodeError(exc)
191 
192         return result
193 
194     def _get_task_meta_for(self, task_id):
195         """Get task meta-data for a task by id."""
196         obj = self.collection.find_one({'_id': task_id})
197         if obj:
198             return self.meta_from_decoded({
199                 'task_id': obj['_id'],
200                 'status': obj['status'],
201                 'result': self.decode(obj['result']),
202                 'date_done': obj['date_done'],
203                 'traceback': self.decode(obj['traceback']),
204                 'children': self.decode(obj['children']),
205             })
206         return {'status': states.PENDING, 'result': None}
207 
208     def _save_group(self, group_id, result):
209         """Save the group result."""
210         self.group_collection.save({
211             '_id': group_id,
212             'result': self.encode([i.id for i in result]),
213             'date_done': datetime.utcnow(),
214         })
215         return result
216 
217     def _restore_group(self, group_id):
218         """Get the result for a group by id."""
219         obj = self.group_collection.find_one({'_id': group_id})
220         if obj:
221             return {
222                 'task_id': obj['_id'],
223                 'date_done': obj['date_done'],
224                 'result': [
225                     self.app.AsyncResult(task)
226                     for task in self.decode(obj['result'])
227                 ],
228             }
229 
230     def _delete_group(self, group_id):
231         """Delete a group by id."""
232         self.group_collection.remove({'_id': group_id})
233 
234     def _forget(self, task_id):
235         """Remove result from MongoDB.
236 
237         Raises:
238             pymongo.exceptions.OperationsError:
239                 if the task_id could not be removed.
240         """
241         # By using safe=True, this will wait until it receives a response from
242         # the server.  Likewise, it will raise an OperationsError if the
243         # response was unable to be completed.
244         self.collection.remove({'_id': task_id})
245 
246     def cleanup(self):
247         """Delete expired meta-data."""
248         self.collection.remove(
249             {'date_done': {'$lt': self.app.now() - self.expires_delta}},
250         )
251         self.group_collection.remove(
252             {'date_done': {'$lt': self.app.now() - self.expires_delta}},
253         )
254 
255     def __reduce__(self, args=(), kwargs={}):
256         return super(MongoBackend, self).__reduce__(
257             args, dict(kwargs, expires=self.expires, url=self.url))
258 
259     def _get_database(self):
260         conn = self._get_connection()
261         db = conn[self.database_name]
262         if self.user and self.password:
263             if not db.authenticate(self.user, self.password):
264                 raise ImproperlyConfigured(
265                     'Invalid MongoDB username or password.')
266         return db
267 
268     @cached_property
269     def database(self):
270         """Get database from MongoDB connection.
271 
272         performs authentication if necessary.
273         """
274         return self._get_database()
275 
276     @cached_property
277     def collection(self):
278         """Get the meta-data task collection."""
279         collection = self.database[self.taskmeta_collection]
280 
281         # Ensure an index on date_done is there, if not process the index
282         # in the background.  Once completed cleanup will be much faster
283         collection.ensure_index('date_done', background='true')
284         return collection
285 
286     @cached_property
287     def group_collection(self):
288         """Get the meta-data task collection."""
289         collection = self.database[self.groupmeta_collection]
290 
291         # Ensure an index on date_done is there, if not process the index
292         # in the background.  Once completed cleanup will be much faster
293         collection.ensure_index('date_done', background='true')
294         return collection
295 
296     @cached_property
297     def expires_delta(self):
298         return timedelta(seconds=self.expires)
299 
300     def as_uri(self, include_password=False):
301         """Return the backend as an URI.
302 
303         Arguments:
304             include_password (bool): Password censored if disabled.
305         """
306         if not self.url:
307             return 'mongodb://'
308         if include_password:
309             return self.url
310 
311         if ',' not in self.url:
312             return maybe_sanitize_url(self.url)
313 
314         uri1, remainder = self.url.split(',', 1)
315         return ','.join([maybe_sanitize_url(uri1), remainder])
316
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/celery/backends/mongodb.py b/celery/backends/mongodb.py
--- a/celery/backends/mongodb.py
+++ b/celery/backends/mongodb.py
@@ -6,7 +6,7 @@
 
 from kombu.exceptions import EncodeError
 from kombu.utils.objects import cached_property
-from kombu.utils.url import maybe_sanitize_url
+from kombu.utils.url import maybe_sanitize_url, urlparse
 
 from celery import states
 from celery.exceptions import ImproperlyConfigured
@@ -75,8 +75,7 @@
 
         # update conf with mongo uri data, only if uri was given
         if self.url:
-            if self.url == 'mongodb://':
-                self.url += 'localhost'
+            self.url = self._ensure_mongodb_uri_compliance(self.url)
 
             uri_data = pymongo.uri_parser.parse_uri(self.url)
             # build the hosts list to create a mongo connection
@@ -120,6 +119,17 @@
             self.options.update(config.pop('options', {}))
             self.options.update(config)
 
+    @staticmethod
+    def _ensure_mongodb_uri_compliance(url):
+        parsed_url = urlparse(url)
+        if not parsed_url.scheme.startswith('mongodb'):
+            url = 'mongodb+{}'.format(url)
+
+        if url == 'mongodb://':
+            url += 'localhost'
+
+        return url
+
     def _prepare_client_options(self):
         if pymongo.version_tuple >= (3,):
             return {'maxPoolSize': self.max_pool_size}

{"golden_diff": "diff --git a/celery/backends/mongodb.py b/celery/backends/mongodb.py\n--- a/celery/backends/mongodb.py\n+++ b/celery/backends/mongodb.py\n@@ -6,7 +6,7 @@\n \n from kombu.exceptions import EncodeError\n from kombu.utils.objects import cached_property\n-from kombu.utils.url import maybe_sanitize_url\n+from kombu.utils.url import maybe_sanitize_url, urlparse\n \n from celery import states\n from celery.exceptions import ImproperlyConfigured\n@@ -75,8 +75,7 @@\n \n         # update conf with mongo uri data, only if uri was given\n         if self.url:\n-            if self.url == 'mongodb://':\n-                self.url += 'localhost'\n+            self.url = self._ensure_mongodb_uri_compliance(self.url)\n \n             uri_data = pymongo.uri_parser.parse_uri(self.url)\n             # build the hosts list to create a mongo connection\n@@ -120,6 +119,17 @@\n             self.options.update(config.pop('options', {}))\n             self.options.update(config)\n \n+    @staticmethod\n+    def _ensure_mongodb_uri_compliance(url):\n+        parsed_url = urlparse(url)\n+        if not parsed_url.scheme.startswith('mongodb'):\n+            url = 'mongodb+{}'.format(url)\n+\n+        if url == 'mongodb://':\n+            url += 'localhost'\n+\n+        return url\n+\n     def _prepare_client_options(self):\n         if pymongo.version_tuple >= (3,):\n             return {'maxPoolSize': self.max_pool_size}\n", "issue": "MongoDB backend does not support mongodb+srv:// URL's\n## Checklist\r\n\r\nhttps://github.com/celery/celery/blob/master/celery/backends/mongodb.py#L143-L146\r\n\r\n## Steps to reproduce\r\n\r\nSet the MongoDB URL to one like:\r\n\r\n```mongodb+srv://mongo.private.corp.example.com/celery?ssl=false```\r\n\r\n## Expected behavior\r\n\r\nThis works.\r\n\r\n## Actual behavior\r\n\r\nThis fails because the URL parsing does not match on `mongodb+srv://` instead trying to match `mongodb://` only.\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"MongoDB result store backend.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nfrom datetime import datetime, timedelta\n\nfrom kombu.exceptions import EncodeError\nfrom kombu.utils.objects import cached_property\nfrom kombu.utils.url import maybe_sanitize_url\n\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.five import items, string_t\n\nfrom .base import BaseBackend\n\ntry:\n    import pymongo\nexcept ImportError:  # pragma: no cover\n    pymongo = None   # noqa\n\nif pymongo:\n    try:\n        from bson.binary import Binary\n    except ImportError:                     # pragma: no cover\n        from pymongo.binary import Binary   # noqa\n    from pymongo.errors import InvalidDocument  # noqa\nelse:                                       # pragma: no cover\n    Binary = None                           # noqa\n\n    class InvalidDocument(Exception):       # noqa\n        pass\n\n__all__ = ('MongoBackend',)\n\nBINARY_CODECS = frozenset(['pickle', 'msgpack'])\n\n\nclass MongoBackend(BaseBackend):\n    \"\"\"MongoDB result backend.\n\n    Raises:\n        celery.exceptions.ImproperlyConfigured:\n            if module :pypi:`pymongo` is not available.\n    \"\"\"\n\n    mongo_host = None\n    host = 'localhost'\n    port = 27017\n    user = None\n    password = None\n    database_name = 'celery'\n    taskmeta_collection = 'celery_taskmeta'\n    groupmeta_collection = 'celery_groupmeta'\n    max_pool_size = 10\n    options = None\n\n    supports_autoexpire = False\n\n    _connection = None\n\n    def __init__(self, app=None, **kwargs):\n        self.options = {}\n\n        super(MongoBackend, self).__init__(app, **kwargs)\n\n        if not pymongo:\n            raise ImproperlyConfigured(\n                'You need to install the pymongo library to use the '\n                'MongoDB backend.')\n\n        # Set option defaults\n        for key, value in items(self._prepare_client_options()):\n            self.options.setdefault(key, value)\n\n        # update conf with mongo uri data, only if uri was given\n        if self.url:\n            if self.url == 'mongodb://':\n                self.url += 'localhost'\n\n            uri_data = pymongo.uri_parser.parse_uri(self.url)\n            # build the hosts list to create a mongo connection\n            hostslist = [\n                '{0}:{1}'.format(x[0], x[1]) for x in uri_data['nodelist']\n            ]\n            self.user = uri_data['username']\n            self.password = uri_data['password']\n            self.mongo_host = hostslist\n            if uri_data['database']:\n                # if no database is provided in the uri, use default\n                self.database_name = uri_data['database']\n\n            self.options.update(uri_data['options'])\n\n        # update conf with specific settings\n        config = self.app.conf.get('mongodb_backend_settings')\n        if config is not None:\n            if not isinstance(config, dict):\n                raise ImproperlyConfigured(\n                    'MongoDB backend settings should be grouped in a dict')\n            config = dict(config)  # don't modify original\n\n            if 'host' in config or 'port' in config:\n                # these should take over uri conf\n                self.mongo_host = None\n\n            self.host = config.pop('host', self.host)\n            self.port = config.pop('port', self.port)\n            self.mongo_host = config.pop('mongo_host', self.mongo_host)\n            self.user = config.pop('user', self.user)\n            self.password = config.pop('password', self.password)\n            self.database_name = config.pop('database', self.database_name)\n            self.taskmeta_collection = config.pop(\n                'taskmeta_collection', self.taskmeta_collection,\n            )\n            self.groupmeta_collection = config.pop(\n                'groupmeta_collection', self.groupmeta_collection,\n            )\n\n            self.options.update(config.pop('options', {}))\n            self.options.update(config)\n\n    def _prepare_client_options(self):\n        if pymongo.version_tuple >= (3,):\n            return {'maxPoolSize': self.max_pool_size}\n        else:  # pragma: no cover\n            return {'max_pool_size': self.max_pool_size,\n                    'auto_start_request': False}\n\n    def _get_connection(self):\n        \"\"\"Connect to the MongoDB server.\"\"\"\n        if self._connection is None:\n            from pymongo import MongoClient\n\n            host = self.mongo_host\n            if not host:\n                # The first pymongo.Connection() argument (host) can be\n                # a list of ['host:port'] elements or a mongodb connection\n                # URI.  If this is the case, don't use self.port\n                # but let pymongo get the port(s) from the URI instead.\n                # This enables the use of replica sets and sharding.\n                # See pymongo.Connection() for more info.\n                host = self.host\n                if isinstance(host, string_t) \\\n                   and not host.startswith('mongodb://'):\n                    host = 'mongodb://{0}:{1}'.format(host, self.port)\n            # don't change self.options\n            conf = dict(self.options)\n            conf['host'] = host\n\n            self._connection = MongoClient(**conf)\n\n        return self._connection\n\n    def encode(self, data):\n        if self.serializer == 'bson':\n            # mongodb handles serialization\n            return data\n        payload = super(MongoBackend, self).encode(data)\n\n        # serializer which are in a unsupported format (pickle/binary)\n        if self.serializer in BINARY_CODECS:\n            payload = Binary(payload)\n        return payload\n\n    def decode(self, data):\n        if self.serializer == 'bson':\n            return data\n        return super(MongoBackend, self).decode(data)\n\n    def _store_result(self, task_id, result, state,\n                      traceback=None, request=None, **kwargs):\n        \"\"\"Store return value and state of an executed task.\"\"\"\n        meta = {\n            '_id': task_id,\n            'status': state,\n            'result': self.encode(result),\n            'date_done': datetime.utcnow(),\n            'traceback': self.encode(traceback),\n            'children': self.encode(\n                self.current_task_children(request),\n            ),\n        }\n        if request and getattr(request, 'parent_id', None):\n            meta['parent_id'] = request.parent_id\n\n        try:\n            self.collection.save(meta)\n        except InvalidDocument as exc:\n            raise EncodeError(exc)\n\n        return result\n\n    def _get_task_meta_for(self, task_id):\n        \"\"\"Get task meta-data for a task by id.\"\"\"\n        obj = self.collection.find_one({'_id': task_id})\n        if obj:\n            return self.meta_from_decoded({\n                'task_id': obj['_id'],\n                'status': obj['status'],\n                'result': self.decode(obj['result']),\n                'date_done': obj['date_done'],\n                'traceback': self.decode(obj['traceback']),\n                'children': self.decode(obj['children']),\n            })\n        return {'status': states.PENDING, 'result': None}\n\n    def _save_group(self, group_id, result):\n        \"\"\"Save the group result.\"\"\"\n        self.group_collection.save({\n            '_id': group_id,\n            'result': self.encode([i.id for i in result]),\n            'date_done': datetime.utcnow(),\n        })\n        return result\n\n    def _restore_group(self, group_id):\n        \"\"\"Get the result for a group by id.\"\"\"\n        obj = self.group_collection.find_one({'_id': group_id})\n        if obj:\n            return {\n                'task_id': obj['_id'],\n                'date_done': obj['date_done'],\n                'result': [\n                    self.app.AsyncResult(task)\n                    for task in self.decode(obj['result'])\n                ],\n            }\n\n    def _delete_group(self, group_id):\n        \"\"\"Delete a group by id.\"\"\"\n        self.group_collection.remove({'_id': group_id})\n\n    def _forget(self, task_id):\n        \"\"\"Remove result from MongoDB.\n\n        Raises:\n            pymongo.exceptions.OperationsError:\n                if the task_id could not be removed.\n        \"\"\"\n        # By using safe=True, this will wait until it receives a response from\n        # the server.  Likewise, it will raise an OperationsError if the\n        # response was unable to be completed.\n        self.collection.remove({'_id': task_id})\n\n    def cleanup(self):\n        \"\"\"Delete expired meta-data.\"\"\"\n        self.collection.remove(\n            {'date_done': {'$lt': self.app.now() - self.expires_delta}},\n        )\n        self.group_collection.remove(\n            {'date_done': {'$lt': self.app.now() - self.expires_delta}},\n        )\n\n    def __reduce__(self, args=(), kwargs={}):\n        return super(MongoBackend, self).__reduce__(\n            args, dict(kwargs, expires=self.expires, url=self.url))\n\n    def _get_database(self):\n        conn = self._get_connection()\n        db = conn[self.database_name]\n        if self.user and self.password:\n            if not db.authenticate(self.user, self.password):\n                raise ImproperlyConfigured(\n                    'Invalid MongoDB username or password.')\n        return db\n\n    @cached_property\n    def database(self):\n        \"\"\"Get database from MongoDB connection.\n\n        performs authentication if necessary.\n        \"\"\"\n        return self._get_database()\n\n    @cached_property\n    def collection(self):\n        \"\"\"Get the meta-data task collection.\"\"\"\n        collection = self.database[self.taskmeta_collection]\n\n        # Ensure an index on date_done is there, if not process the index\n        # in the background.  Once completed cleanup will be much faster\n        collection.ensure_index('date_done', background='true')\n        return collection\n\n    @cached_property\n    def group_collection(self):\n        \"\"\"Get the meta-data task collection.\"\"\"\n        collection = self.database[self.groupmeta_collection]\n\n        # Ensure an index on date_done is there, if not process the index\n        # in the background.  Once completed cleanup will be much faster\n        collection.ensure_index('date_done', background='true')\n        return collection\n\n    @cached_property\n    def expires_delta(self):\n        return timedelta(seconds=self.expires)\n\n    def as_uri(self, include_password=False):\n        \"\"\"Return the backend as an URI.\n\n        Arguments:\n            include_password (bool): Password censored if disabled.\n        \"\"\"\n        if not self.url:\n            return 'mongodb://'\n        if include_password:\n            return self.url\n\n        if ',' not in self.url:\n            return maybe_sanitize_url(self.url)\n\n        uri1, remainder = self.url.split(',', 1)\n        return ','.join([maybe_sanitize_url(uri1), remainder])\n", "path": "celery/backends/mongodb.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"MongoDB result store backend.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nfrom datetime import datetime, timedelta\n\nfrom kombu.exceptions import EncodeError\nfrom kombu.utils.objects import cached_property\nfrom kombu.utils.url import maybe_sanitize_url, urlparse\n\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.five import items, string_t\n\nfrom .base import BaseBackend\n\ntry:\n    import pymongo\nexcept ImportError:  # pragma: no cover\n    pymongo = None   # noqa\n\nif pymongo:\n    try:\n        from bson.binary import Binary\n    except ImportError:                     # pragma: no cover\n        from pymongo.binary import Binary   # noqa\n    from pymongo.errors import InvalidDocument  # noqa\nelse:                                       # pragma: no cover\n    Binary = None                           # noqa\n\n    class InvalidDocument(Exception):       # noqa\n        pass\n\n__all__ = ('MongoBackend',)\n\nBINARY_CODECS = frozenset(['pickle', 'msgpack'])\n\n\nclass MongoBackend(BaseBackend):\n    \"\"\"MongoDB result backend.\n\n    Raises:\n        celery.exceptions.ImproperlyConfigured:\n            if module :pypi:`pymongo` is not available.\n    \"\"\"\n\n    mongo_host = None\n    host = 'localhost'\n    port = 27017\n    user = None\n    password = None\n    database_name = 'celery'\n    taskmeta_collection = 'celery_taskmeta'\n    groupmeta_collection = 'celery_groupmeta'\n    max_pool_size = 10\n    options = None\n\n    supports_autoexpire = False\n\n    _connection = None\n\n    def __init__(self, app=None, **kwargs):\n        self.options = {}\n\n        super(MongoBackend, self).__init__(app, **kwargs)\n\n        if not pymongo:\n            raise ImproperlyConfigured(\n                'You need to install the pymongo library to use the '\n                'MongoDB backend.')\n\n        # Set option defaults\n        for key, value in items(self._prepare_client_options()):\n            self.options.setdefault(key, value)\n\n        # update conf with mongo uri data, only if uri was given\n        if self.url:\n            self.url = self._ensure_mongodb_uri_compliance(self.url)\n\n            uri_data = pymongo.uri_parser.parse_uri(self.url)\n            # build the hosts list to create a mongo connection\n            hostslist = [\n                '{0}:{1}'.format(x[0], x[1]) for x in uri_data['nodelist']\n            ]\n            self.user = uri_data['username']\n            self.password = uri_data['password']\n            self.mongo_host = hostslist\n            if uri_data['database']:\n                # if no database is provided in the uri, use default\n                self.database_name = uri_data['database']\n\n            self.options.update(uri_data['options'])\n\n        # update conf with specific settings\n        config = self.app.conf.get('mongodb_backend_settings')\n        if config is not None:\n            if not isinstance(config, dict):\n                raise ImproperlyConfigured(\n                    'MongoDB backend settings should be grouped in a dict')\n            config = dict(config)  # don't modify original\n\n            if 'host' in config or 'port' in config:\n                # these should take over uri conf\n                self.mongo_host = None\n\n            self.host = config.pop('host', self.host)\n            self.port = config.pop('port', self.port)\n            self.mongo_host = config.pop('mongo_host', self.mongo_host)\n            self.user = config.pop('user', self.user)\n            self.password = config.pop('password', self.password)\n            self.database_name = config.pop('database', self.database_name)\n            self.taskmeta_collection = config.pop(\n                'taskmeta_collection', self.taskmeta_collection,\n            )\n            self.groupmeta_collection = config.pop(\n                'groupmeta_collection', self.groupmeta_collection,\n            )\n\n            self.options.update(config.pop('options', {}))\n            self.options.update(config)\n\n    @staticmethod\n    def _ensure_mongodb_uri_compliance(url):\n        parsed_url = urlparse(url)\n        if not parsed_url.scheme.startswith('mongodb'):\n            url = 'mongodb+{}'.format(url)\n\n        if url == 'mongodb://':\n            url += 'localhost'\n\n        return url\n\n    def _prepare_client_options(self):\n        if pymongo.version_tuple >= (3,):\n            return {'maxPoolSize': self.max_pool_size}\n        else:  # pragma: no cover\n            return {'max_pool_size': self.max_pool_size,\n                    'auto_start_request': False}\n\n    def _get_connection(self):\n        \"\"\"Connect to the MongoDB server.\"\"\"\n        if self._connection is None:\n            from pymongo import MongoClient\n\n            host = self.mongo_host\n            if not host:\n                # The first pymongo.Connection() argument (host) can be\n                # a list of ['host:port'] elements or a mongodb connection\n                # URI.  If this is the case, don't use self.port\n                # but let pymongo get the port(s) from the URI instead.\n                # This enables the use of replica sets and sharding.\n                # See pymongo.Connection() for more info.\n                host = self.host\n                if isinstance(host, string_t) \\\n                   and not host.startswith('mongodb://'):\n                    host = 'mongodb://{0}:{1}'.format(host, self.port)\n            # don't change self.options\n            conf = dict(self.options)\n            conf['host'] = host\n\n            self._connection = MongoClient(**conf)\n\n        return self._connection\n\n    def encode(self, data):\n        if self.serializer == 'bson':\n            # mongodb handles serialization\n            return data\n        payload = super(MongoBackend, self).encode(data)\n\n        # serializer which are in a unsupported format (pickle/binary)\n        if self.serializer in BINARY_CODECS:\n            payload = Binary(payload)\n        return payload\n\n    def decode(self, data):\n        if self.serializer == 'bson':\n            return data\n        return super(MongoBackend, self).decode(data)\n\n    def _store_result(self, task_id, result, state,\n                      traceback=None, request=None, **kwargs):\n        \"\"\"Store return value and state of an executed task.\"\"\"\n        meta = {\n            '_id': task_id,\n            'status': state,\n            'result': self.encode(result),\n            'date_done': datetime.utcnow(),\n            'traceback': self.encode(traceback),\n            'children': self.encode(\n                self.current_task_children(request),\n            ),\n        }\n        if request and getattr(request, 'parent_id', None):\n            meta['parent_id'] = request.parent_id\n\n        try:\n            self.collection.save(meta)\n        except InvalidDocument as exc:\n            raise EncodeError(exc)\n\n        return result\n\n    def _get_task_meta_for(self, task_id):\n        \"\"\"Get task meta-data for a task by id.\"\"\"\n        obj = self.collection.find_one({'_id': task_id})\n        if obj:\n            return self.meta_from_decoded({\n                'task_id': obj['_id'],\n                'status': obj['status'],\n                'result': self.decode(obj['result']),\n                'date_done': obj['date_done'],\n                'traceback': self.decode(obj['traceback']),\n                'children': self.decode(obj['children']),\n            })\n        return {'status': states.PENDING, 'result': None}\n\n    def _save_group(self, group_id, result):\n        \"\"\"Save the group result.\"\"\"\n        self.group_collection.save({\n            '_id': group_id,\n            'result': self.encode([i.id for i in result]),\n            'date_done': datetime.utcnow(),\n        })\n        return result\n\n    def _restore_group(self, group_id):\n        \"\"\"Get the result for a group by id.\"\"\"\n        obj = self.group_collection.find_one({'_id': group_id})\n        if obj:\n            return {\n                'task_id': obj['_id'],\n                'date_done': obj['date_done'],\n                'result': [\n                    self.app.AsyncResult(task)\n                    for task in self.decode(obj['result'])\n                ],\n            }\n\n    def _delete_group(self, group_id):\n        \"\"\"Delete a group by id.\"\"\"\n        self.group_collection.remove({'_id': group_id})\n\n    def _forget(self, task_id):\n        \"\"\"Remove result from MongoDB.\n\n        Raises:\n            pymongo.exceptions.OperationsError:\n                if the task_id could not be removed.\n        \"\"\"\n        # By using safe=True, this will wait until it receives a response from\n        # the server.  Likewise, it will raise an OperationsError if the\n        # response was unable to be completed.\n        self.collection.remove({'_id': task_id})\n\n    def cleanup(self):\n        \"\"\"Delete expired meta-data.\"\"\"\n        self.collection.remove(\n            {'date_done': {'$lt': self.app.now() - self.expires_delta}},\n        )\n        self.group_collection.remove(\n            {'date_done': {'$lt': self.app.now() - self.expires_delta}},\n        )\n\n    def __reduce__(self, args=(), kwargs={}):\n        return super(MongoBackend, self).__reduce__(\n            args, dict(kwargs, expires=self.expires, url=self.url))\n\n    def _get_database(self):\n        conn = self._get_connection()\n        db = conn[self.database_name]\n        if self.user and self.password:\n            if not db.authenticate(self.user, self.password):\n                raise ImproperlyConfigured(\n                    'Invalid MongoDB username or password.')\n        return db\n\n    @cached_property\n    def database(self):\n        \"\"\"Get database from MongoDB connection.\n\n        performs authentication if necessary.\n        \"\"\"\n        return self._get_database()\n\n    @cached_property\n    def collection(self):\n        \"\"\"Get the meta-data task collection.\"\"\"\n        collection = self.database[self.taskmeta_collection]\n\n        # Ensure an index on date_done is there, if not process the index\n        # in the background.  Once completed cleanup will be much faster\n        collection.ensure_index('date_done', background='true')\n        return collection\n\n    @cached_property\n    def group_collection(self):\n        \"\"\"Get the meta-data task collection.\"\"\"\n        collection = self.database[self.groupmeta_collection]\n\n        # Ensure an index on date_done is there, if not process the index\n        # in the background.  Once completed cleanup will be much faster\n        collection.ensure_index('date_done', background='true')\n        return collection\n\n    @cached_property\n    def expires_delta(self):\n        return timedelta(seconds=self.expires)\n\n    def as_uri(self, include_password=False):\n        \"\"\"Return the backend as an URI.\n\n        Arguments:\n            include_password (bool): Password censored if disabled.\n        \"\"\"\n        if not self.url:\n            return 'mongodb://'\n        if include_password:\n            return self.url\n\n        if ',' not in self.url:\n            return maybe_sanitize_url(self.url)\n\n        uri1, remainder = self.url.split(',', 1)\n        return ','.join([maybe_sanitize_url(uri1), remainder])\n", "path": "celery/backends/mongodb.py"}]}