We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Celery does not consider authSource on mongodb backend URLs
Version: Celery 4.0.2 (from looking at the changes since then it seems there is no change addressing this issue here: https://github.com/celery/celery/commits/master/celery/backends/mongodb.py )

(Edit) Confirmed with the following versions as well:
amqp==2.2.2
billiard==3.5.0.3
celery==4.1.0
kombu==4.1.0
pymongo==3.6.0

Celery Report

software -> celery:4.0.2 (latentcall) kombu:4.0.2 py:2.7.8
            billiard:3.5.0.3 py-amqp:2.2.1
platform -> system:Linux arch:64bit, ELF imp:CPython
loader   -> celery.loaders.default.Loader
settings -> transport:amqp results:disabled


## Steps to reproduce

Give Celery a Backend URL pointing to a MongoDB instance with authentication and username/password (user/pwd set on the Admin DB by default) in the format:

mongodb://user:pass@your-server/your_db?authSource=admin

(Please see http://api.mongodb.com/python/current/examples/authentication.html#default-database-and-authsource and http://api.mongodb.com/python/current/api/pymongo/mongo_client.html?highlight=authsource )

## Expected behavior

Celery authenticates the user in the admin database (this is the same as passing --authenticationDatabase to the mongo client or the same url to MongoClient)

## Actual behavior

Celery tries to authenticate the user on the your_db database (failing to authenticate)

## Workaround (not recommended)

Change the db on the URL to /admin (this db shouldn't be used to store arbitrary data normally)

--- 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, urlparse
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             self.url = self._ensure_mongodb_uri_compliance(self.url)
79 
80             uri_data = pymongo.uri_parser.parse_uri(self.url)
81             # build the hosts list to create a mongo connection
82             hostslist = [
83                 '{0}:{1}'.format(x[0], x[1]) for x in uri_data['nodelist']
84             ]
85             self.user = uri_data['username']
86             self.password = uri_data['password']
87             self.mongo_host = hostslist
88             if uri_data['database']:
89                 # if no database is provided in the uri, use default
90                 self.database_name = uri_data['database']
91 
92             self.options.update(uri_data['options'])
93 
94         # update conf with specific settings
95         config = self.app.conf.get('mongodb_backend_settings')
96         if config is not None:
97             if not isinstance(config, dict):
98                 raise ImproperlyConfigured(
99                     'MongoDB backend settings should be grouped in a dict')
100             config = dict(config)  # don't modify original
101 
102             if 'host' in config or 'port' in config:
103                 # these should take over uri conf
104                 self.mongo_host = None
105 
106             self.host = config.pop('host', self.host)
107             self.port = config.pop('port', self.port)
108             self.mongo_host = config.pop('mongo_host', self.mongo_host)
109             self.user = config.pop('user', self.user)
110             self.password = config.pop('password', self.password)
111             self.database_name = config.pop('database', self.database_name)
112             self.taskmeta_collection = config.pop(
113                 'taskmeta_collection', self.taskmeta_collection,
114             )
115             self.groupmeta_collection = config.pop(
116                 'groupmeta_collection', self.groupmeta_collection,
117             )
118 
119             self.options.update(config.pop('options', {}))
120             self.options.update(config)
121 
122     @staticmethod
123     def _ensure_mongodb_uri_compliance(url):
124         parsed_url = urlparse(url)
125         if not parsed_url.scheme.startswith('mongodb'):
126             url = 'mongodb+{}'.format(url)
127 
128         if url == 'mongodb://':
129             url += 'localhost'
130 
131         return url
132 
133     def _prepare_client_options(self):
134         if pymongo.version_tuple >= (3,):
135             return {'maxPoolSize': self.max_pool_size}
136         else:  # pragma: no cover
137             return {'max_pool_size': self.max_pool_size,
138                     'auto_start_request': False}
139 
140     def _get_connection(self):
141         """Connect to the MongoDB server."""
142         if self._connection is None:
143             from pymongo import MongoClient
144 
145             host = self.mongo_host
146             if not host:
147                 # The first pymongo.Connection() argument (host) can be
148                 # a list of ['host:port'] elements or a mongodb connection
149                 # URI.  If this is the case, don't use self.port
150                 # but let pymongo get the port(s) from the URI instead.
151                 # This enables the use of replica sets and sharding.
152                 # See pymongo.Connection() for more info.
153                 host = self.host
154                 if isinstance(host, string_t) \
155                    and not host.startswith('mongodb://'):
156                     host = 'mongodb://{0}:{1}'.format(host, self.port)
157             # don't change self.options
158             conf = dict(self.options)
159             conf['host'] = host
160 
161             self._connection = MongoClient(**conf)
162 
163         return self._connection
164 
165     def encode(self, data):
166         if self.serializer == 'bson':
167             # mongodb handles serialization
168             return data
169         payload = super(MongoBackend, self).encode(data)
170 
171         # serializer which are in a unsupported format (pickle/binary)
172         if self.serializer in BINARY_CODECS:
173             payload = Binary(payload)
174         return payload
175 
176     def decode(self, data):
177         if self.serializer == 'bson':
178             return data
179         return super(MongoBackend, self).decode(data)
180 
181     def _store_result(self, task_id, result, state,
182                       traceback=None, request=None, **kwargs):
183         """Store return value and state of an executed task."""
184         meta = {
185             '_id': task_id,
186             'status': state,
187             'result': self.encode(result),
188             'date_done': datetime.utcnow(),
189             'traceback': self.encode(traceback),
190             'children': self.encode(
191                 self.current_task_children(request),
192             ),
193         }
194         if request and getattr(request, 'parent_id', None):
195             meta['parent_id'] = request.parent_id
196 
197         try:
198             self.collection.replace_one({'_id': task_id}, meta, upsert=True)
199         except InvalidDocument as exc:
200             raise EncodeError(exc)
201 
202         return result
203 
204     def _get_task_meta_for(self, task_id):
205         """Get task meta-data for a task by id."""
206         obj = self.collection.find_one({'_id': task_id})
207         if obj:
208             return self.meta_from_decoded({
209                 'task_id': obj['_id'],
210                 'status': obj['status'],
211                 'result': self.decode(obj['result']),
212                 'date_done': obj['date_done'],
213                 'traceback': self.decode(obj['traceback']),
214                 'children': self.decode(obj['children']),
215             })
216         return {'status': states.PENDING, 'result': None}
217 
218     def _save_group(self, group_id, result):
219         """Save the group result."""
220         meta = {
221             '_id': group_id,
222             'result': self.encode([i.id for i in result]),
223             'date_done': datetime.utcnow(),
224         }
225         self.group_collection.replace_one({'_id': group_id}, meta, upsert=True)
226         return result
227 
228     def _restore_group(self, group_id):
229         """Get the result for a group by id."""
230         obj = self.group_collection.find_one({'_id': group_id})
231         if obj:
232             return {
233                 'task_id': obj['_id'],
234                 'date_done': obj['date_done'],
235                 'result': [
236                     self.app.AsyncResult(task)
237                     for task in self.decode(obj['result'])
238                 ],
239             }
240 
241     def _delete_group(self, group_id):
242         """Delete a group by id."""
243         self.group_collection.delete_one({'_id': group_id})
244 
245     def _forget(self, task_id):
246         """Remove result from MongoDB.
247 
248         Raises:
249             pymongo.exceptions.OperationsError:
250                 if the task_id could not be removed.
251         """
252         # By using safe=True, this will wait until it receives a response from
253         # the server.  Likewise, it will raise an OperationsError if the
254         # response was unable to be completed.
255         self.collection.delete_one({'_id': task_id})
256 
257     def cleanup(self):
258         """Delete expired meta-data."""
259         self.collection.delete_many(
260             {'date_done': {'$lt': self.app.now() - self.expires_delta}},
261         )
262         self.group_collection.delete_many(
263             {'date_done': {'$lt': self.app.now() - self.expires_delta}},
264         )
265 
266     def __reduce__(self, args=(), kwargs={}):
267         return super(MongoBackend, self).__reduce__(
268             args, dict(kwargs, expires=self.expires, url=self.url))
269 
270     def _get_database(self):
271         conn = self._get_connection()
272         db = conn[self.database_name]
273         if self.user and self.password:
274             if not db.authenticate(self.user, self.password):
275                 raise ImproperlyConfigured(
276                     'Invalid MongoDB username or password.')
277         return db
278 
279     @cached_property
280     def database(self):
281         """Get database from MongoDB connection.
282 
283         performs authentication if necessary.
284         """
285         return self._get_database()
286 
287     @cached_property
288     def collection(self):
289         """Get the meta-data task collection."""
290         collection = self.database[self.taskmeta_collection]
291 
292         # Ensure an index on date_done is there, if not process the index
293         # in the background.  Once completed cleanup will be much faster
294         collection.create_index('date_done', background=True)
295         return collection
296 
297     @cached_property
298     def group_collection(self):
299         """Get the meta-data task collection."""
300         collection = self.database[self.groupmeta_collection]
301 
302         # Ensure an index on date_done is there, if not process the index
303         # in the background.  Once completed cleanup will be much faster
304         collection.create_index('date_done', background=True)
305         return collection
306 
307     @cached_property
308     def expires_delta(self):
309         return timedelta(seconds=self.expires)
310 
311     def as_uri(self, include_password=False):
312         """Return the backend as an URI.
313 
314         Arguments:
315             include_password (bool): Password censored if disabled.
316         """
317         if not self.url:
318             return 'mongodb://'
319         if include_password:
320             return self.url
321 
322         if ',' not in self.url:
323             return maybe_sanitize_url(self.url)
324 
325         uri1, remainder = self.url.split(',', 1)
326         return ','.join([maybe_sanitize_url(uri1), remainder])
327
```
--- 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
@@ -271,7 +271,10 @@
         conn = self._get_connection()
         db = conn[self.database_name]
         if self.user and self.password:
-            if not db.authenticate(self.user, self.password):
+            source = self.options.get('authsource', 
+                        self.database_name or 'admin'
+            )
+            if not db.authenticate(self.user, self.password, source=source):
                 raise ImproperlyConfigured(
                     'Invalid MongoDB username or password.')
         return db

{"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@@ -271,7 +271,10 @@\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+            source = self.options.get('authsource', \n+                        self.database_name or 'admin'\n+            )\n+            if not db.authenticate(self.user, self.password, source=source):\n                 raise ImproperlyConfigured(\n                     'Invalid MongoDB username or password.')\n         return db\n", "issue": "Celery does not consider authSource on mongodb backend URLs\nVersion: Celery 4.0.2 (from looking at the changes since then it seems there is no change addressing this issue here: https://github.com/celery/celery/commits/master/celery/backends/mongodb.py )\r\n\r\n(Edit) Confirmed with the following versions as well:\r\namqp==2.2.2\r\nbilliard==3.5.0.3\r\ncelery==4.1.0\r\nkombu==4.1.0\r\npymongo==3.6.0\r\n\r\nCelery Report\r\n\r\nsoftware -> celery:4.0.2 (latentcall) kombu:4.0.2 py:2.7.8\r\n            billiard:3.5.0.3 py-amqp:2.2.1\r\nplatform -> system:Linux arch:64bit, ELF imp:CPython\r\nloader   -> celery.loaders.default.Loader\r\nsettings -> transport:amqp results:disabled\r\n\r\n\r\n## Steps to reproduce\r\n\r\nGive Celery a Backend URL pointing to a MongoDB instance with authentication and username/password (user/pwd set on the Admin DB by default) in the format:\r\n\r\nmongodb://user:pass@your-server/your_db?authSource=admin\r\n\r\n(Please see http://api.mongodb.com/python/current/examples/authentication.html#default-database-and-authsource and http://api.mongodb.com/python/current/api/pymongo/mongo_client.html?highlight=authsource )\r\n\r\n## Expected behavior\r\n\r\nCelery authenticates the user in the admin database (this is the same as passing --authenticationDatabase to the mongo client or the same url to MongoClient)\r\n\r\n## Actual behavior\r\n\r\nCelery tries to authenticate the user on the your_db database (failing to authenticate)\r\n\r\n## Workaround (not recommended)\r\n\r\nChange the db on the URL to /admin (this db shouldn't be used to store arbitrary data normally)\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, 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.replace_one({'_id': task_id}, meta, upsert=True)\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        meta = {\n            '_id': group_id,\n            'result': self.encode([i.id for i in result]),\n            'date_done': datetime.utcnow(),\n        }\n        self.group_collection.replace_one({'_id': group_id}, meta, upsert=True)\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.delete_one({'_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.delete_one({'_id': task_id})\n\n    def cleanup(self):\n        \"\"\"Delete expired meta-data.\"\"\"\n        self.collection.delete_many(\n            {'date_done': {'$lt': self.app.now() - self.expires_delta}},\n        )\n        self.group_collection.delete_many(\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.create_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.create_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.replace_one({'_id': task_id}, meta, upsert=True)\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        meta = {\n            '_id': group_id,\n            'result': self.encode([i.id for i in result]),\n            'date_done': datetime.utcnow(),\n        }\n        self.group_collection.replace_one({'_id': group_id}, meta, upsert=True)\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.delete_one({'_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.delete_one({'_id': task_id})\n\n    def cleanup(self):\n        \"\"\"Delete expired meta-data.\"\"\"\n        self.collection.delete_many(\n            {'date_done': {'$lt': self.app.now() - self.expires_delta}},\n        )\n        self.group_collection.delete_many(\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            source = self.options.get('authsource', \n                        self.database_name or 'admin'\n            )\n            if not db.authenticate(self.user, self.password, source=source):\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.create_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.create_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"}]}