We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
pymongo refuses backend/mongodb.py default settings
`self.options.setdefault('max_pool_size', self.max_pool_size)`
`self.options.setdefault('auto_start_request', False)`

These lines are refusing from `pymongo` and gives `ConfigurationError` exception. By the way pymongo version is 3.0 and for use mongodb as backend i have to remove that lines from mongodb.py. 


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

{"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@@ -78,9 +78,14 @@\n \n         self.url = url\n \n-        # default options\n-        self.options.setdefault('max_pool_size', self.max_pool_size)\n-        self.options.setdefault('auto_start_request', False)\n+\n+        # default options according to pymongo version\n+        if pymongo.version_tuple >= (3,):\n+            self.options.setdefault('maxPoolSize', self.max_pool_size)\n+        else:\n+            self.options.setdefault('max_pool_size', self.max_pool_size)\n+            self.options.setdefault('auto_start_request', False)\n+\n \n         # update conf with mongo uri data, only if uri was given\n         if self.url:\n", "issue": "pymongo refuses backend/mongodb.py default settings\n`self.options.setdefault('max_pool_size', self.max_pool_size)`\n`self.options.setdefault('auto_start_request', False)`\n\nThese lines are refusing from `pymongo` and gives `ConfigurationError` exception. By the way pymongo version is 3.0 and for use mongodb as backend i have to remove that lines from mongodb.py. \n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"\n    celery.backends.mongodb\n    ~~~~~~~~~~~~~~~~~~~~~~~\n\n    MongoDB result store backend.\n\n\"\"\"\nfrom __future__ import absolute_import\n\nfrom datetime import datetime, timedelta\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    InvalidDocument = None                  # noqa\n\nfrom kombu.syn import detect_environment\nfrom kombu.utils import cached_property\nfrom kombu.exceptions import EncodeError\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.five import string_t\n\nfrom .base import BaseBackend\n\n__all__ = ['MongoBackend']\n\n\nclass Bunch(object):\n\n    def __init__(self, **kw):\n        self.__dict__.update(kw)\n\n\nclass MongoBackend(BaseBackend):\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, url=None, **kwargs):\n        \"\"\"Initialize MongoDB backend instance.\n\n        :raises celery.exceptions.ImproperlyConfigured: if\n            module :mod:`pymongo` is not available.\n\n        \"\"\"\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        self.url = url\n\n        # default options\n        self.options.setdefault('max_pool_size', self.max_pool_size)\n        self.options.setdefault('auto_start_request', False)\n\n        # update conf with mongo uri data, only if uri was given\n        if self.url:\n            uri_data = pymongo.uri_parser.parse_uri(self.url)\n            # build the hosts list to create a mongo connection\n            make_host_str = lambda x: \"{0}:{1}\".format(x[0], x[1])\n            hostslist = map(make_host_str, uri_data['nodelist'])\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('CELERY_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)  # do not 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 _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\n                if host == 'mongodb://':\n                    host += 'localhost'\n\n            # don't change self.options\n            conf = dict(self.options)\n            conf['host'] = host\n\n            if detect_environment() != 'default':\n                conf['use_greenlets'] = True\n\n            self._connection = MongoClient(**conf)\n\n        return self._connection\n\n    def process_cleanup(self):\n        if self._connection is not None:\n            # MongoDB connection will be closed automatically when object\n            # goes out of scope\n            del(self.collection)\n            del(self.database)\n            self._connection = None\n\n    def encode(self, data):\n        if self.serializer == 'bson':\n            # mongodb handles serialization\n            return data\n        return super(MongoBackend, self).encode(data)\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, status,\n                      traceback=None, request=None, **kwargs):\n        \"\"\"Store return value and status of an executed task.\"\"\"\n\n        meta = {'_id': task_id,\n                'status': status,\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        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 metadata 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\n        task_ids = [i.id for i in result]\n\n        meta = {'_id': group_id,\n                'result': self.encode(task_ids),\n                'date_done': datetime.utcnow()}\n        self.group_collection.save(meta)\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            tasks = [self.app.AsyncResult(task)\n                     for task in self.decode(obj['result'])]\n\n            return {\n                'task_id': obj['_id'],\n                'result': tasks,\n                'date_done': obj['date_done'],\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        \"\"\"\n        Remove result from MongoDB.\n\n        :raises celery.exceptions.OperationsError: if the task_id could not be\n                                                   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 metadata.\"\"\"\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\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,\n                                   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 and perform authentication\n        if necessary.\"\"\"\n        return self._get_database()\n\n    @cached_property\n    def collection(self):\n        \"\"\"Get the metadata 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 metadata 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", "path": "celery/backends/mongodb.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"\n    celery.backends.mongodb\n    ~~~~~~~~~~~~~~~~~~~~~~~\n\n    MongoDB result store backend.\n\n\"\"\"\nfrom __future__ import absolute_import\n\nfrom datetime import datetime, timedelta\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    InvalidDocument = None                  # noqa\n\nfrom kombu.syn import detect_environment\nfrom kombu.utils import cached_property\nfrom kombu.exceptions import EncodeError\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.five import string_t\n\nfrom .base import BaseBackend\n\n__all__ = ['MongoBackend']\n\n\nclass Bunch(object):\n\n    def __init__(self, **kw):\n        self.__dict__.update(kw)\n\n\nclass MongoBackend(BaseBackend):\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, url=None, **kwargs):\n        \"\"\"Initialize MongoDB backend instance.\n\n        :raises celery.exceptions.ImproperlyConfigured: if\n            module :mod:`pymongo` is not available.\n\n        \"\"\"\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        self.url = url\n\n\n        # default options according to pymongo version\n        if pymongo.version_tuple >= (3,):\n            self.options.setdefault('maxPoolSize', self.max_pool_size)\n        else:\n            self.options.setdefault('max_pool_size', self.max_pool_size)\n            self.options.setdefault('auto_start_request', False)\n\n\n        # update conf with mongo uri data, only if uri was given\n        if self.url:\n            uri_data = pymongo.uri_parser.parse_uri(self.url)\n            # build the hosts list to create a mongo connection\n            make_host_str = lambda x: \"{0}:{1}\".format(x[0], x[1])\n            hostslist = map(make_host_str, uri_data['nodelist'])\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('CELERY_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)  # do not 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 _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\n                if host == 'mongodb://':\n                    host += 'localhost'\n\n            # don't change self.options\n            conf = dict(self.options)\n            conf['host'] = host\n\n            if detect_environment() != 'default':\n                conf['use_greenlets'] = True\n\n            self._connection = MongoClient(**conf)\n\n        return self._connection\n\n    def process_cleanup(self):\n        if self._connection is not None:\n            # MongoDB connection will be closed automatically when object\n            # goes out of scope\n            del(self.collection)\n            del(self.database)\n            self._connection = None\n\n    def encode(self, data):\n        if self.serializer == 'bson':\n            # mongodb handles serialization\n            return data\n        return super(MongoBackend, self).encode(data)\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, status,\n                      traceback=None, request=None, **kwargs):\n        \"\"\"Store return value and status of an executed task.\"\"\"\n\n        meta = {'_id': task_id,\n                'status': status,\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        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 metadata 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\n        task_ids = [i.id for i in result]\n\n        meta = {'_id': group_id,\n                'result': self.encode(task_ids),\n                'date_done': datetime.utcnow()}\n        self.group_collection.save(meta)\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            tasks = [self.app.AsyncResult(task)\n                     for task in self.decode(obj['result'])]\n\n            return {\n                'task_id': obj['_id'],\n                'result': tasks,\n                'date_done': obj['date_done'],\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        \"\"\"\n        Remove result from MongoDB.\n\n        :raises celery.exceptions.OperationsError: if the task_id could not be\n                                                   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 metadata.\"\"\"\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\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,\n                                   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 and perform authentication\n        if necessary.\"\"\"\n        return self._get_database()\n\n    @cached_property\n    def collection(self):\n        \"\"\"Get the metadata 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 metadata 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", "path": "celery/backends/mongodb.py"}]}