We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
mongodb result backend with pickle serialisation broken in 4.0.0
Just upgraded to celery 4.0.0. Mongodb result backend with pickle no longer works with the exception at the end of this message. 

This is because commit https://github.com/celery/celery/commit/639b40f6308267312a1030bb3d6ac5805069510a removes the **essential** Binary() wrapper from the data written to mongo. As a result it writes raw binary data to mongo as a string, which then rejects it.

I'm using settings:
CELERY_ACCEPT_CONTENT = ['pickle']
CELERY_TASK_SERIALIZER = 'pickle'
CELERY_RESULT_SERIALIZER = 'pickle'

Exception seen:
> [2016-11-10 16:01:02,118: ERROR/MainProcess] Pool callback raised exception: InvalidStringData("strings in documents must be valid UTF-8: '\\x80\\x02N.'",)
> Traceback (most recent call last):
>   File "/usr/lib/python2.7/site-packages/billiard/pool.py", line 1748, in safe_apply_callback
>     fun(*args, **kwargs)
>   File "/usr/lib/python2.7/site-packages/celery/worker/request.py", line 366, in on_failure
>     self.id, exc, request=self, store_result=self.store_errors,
>   File "/usr/lib/python2.7/site-packages/celery/backends/base.py", line 163, in mark_as_failure
>     traceback=traceback, request=request)
>   File "/usr/lib/python2.7/site-packages/celery/backends/base.py", line 309, in store_result
>     request=request, **kwargs)
>   File "/usr/lib/python2.7/site-packages/celery/backends/mongodb.py", line 175, in _store_result
>     self.collection.save(meta)
>   File "/usr/lib/python2.7/site-packages/pymongo/collection.py", line 2192, in save
>     check_keys, False, manipulate, write_concern)
>   File "/usr/lib/python2.7/site-packages/pymongo/collection.py", line 715, in _update
>     codec_options=self.__write_response_codec_options).copy()
>   File "/usr/lib/python2.7/site-packages/pymongo/pool.py", line 244, in command
>     self._raise_connection_failure(error)
>   File "/usr/lib/python2.7/site-packages/pymongo/pool.py", line 372, in _raise_connection_failure
>     raise error
> InvalidStringData: strings in documents must be valid UTF-8: '\x80\x02N.'


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

{"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@@ -29,6 +29,8 @@\n \n __all__ = ['MongoBackend']\n \n+BINARY_CODECS = frozenset(['pickle', 'msgpack'])\n+\n \n class MongoBackend(BaseBackend):\n     \"\"\"MongoDB result backend.\n@@ -150,7 +152,12 @@\n         if self.serializer == 'bson':\n             # mongodb handles serialization\n             return data\n-        return super(MongoBackend, self).encode(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", "issue": "mongodb result backend with pickle serialisation broken in 4.0.0\nJust upgraded to celery 4.0.0. Mongodb result backend with pickle no longer works with the exception at the end of this message. \r\n\r\nThis is because commit https://github.com/celery/celery/commit/639b40f6308267312a1030bb3d6ac5805069510a removes the **essential** Binary() wrapper from the data written to mongo. As a result it writes raw binary data to mongo as a string, which then rejects it.\r\n\r\nI'm using settings:\r\nCELERY_ACCEPT_CONTENT = ['pickle']\r\nCELERY_TASK_SERIALIZER = 'pickle'\r\nCELERY_RESULT_SERIALIZER = 'pickle'\r\n\r\nException seen:\r\n> [2016-11-10 16:01:02,118: ERROR/MainProcess] Pool callback raised exception: InvalidStringData(\"strings in documents must be valid UTF-8: '\\\\x80\\\\x02N.'\",)\r\n> Traceback (most recent call last):\r\n>   File \"/usr/lib/python2.7/site-packages/billiard/pool.py\", line 1748, in safe_apply_callback\r\n>     fun(*args, **kwargs)\r\n>   File \"/usr/lib/python2.7/site-packages/celery/worker/request.py\", line 366, in on_failure\r\n>     self.id, exc, request=self, store_result=self.store_errors,\r\n>   File \"/usr/lib/python2.7/site-packages/celery/backends/base.py\", line 163, in mark_as_failure\r\n>     traceback=traceback, request=request)\r\n>   File \"/usr/lib/python2.7/site-packages/celery/backends/base.py\", line 309, in store_result\r\n>     request=request, **kwargs)\r\n>   File \"/usr/lib/python2.7/site-packages/celery/backends/mongodb.py\", line 175, in _store_result\r\n>     self.collection.save(meta)\r\n>   File \"/usr/lib/python2.7/site-packages/pymongo/collection.py\", line 2192, in save\r\n>     check_keys, False, manipulate, write_concern)\r\n>   File \"/usr/lib/python2.7/site-packages/pymongo/collection.py\", line 715, in _update\r\n>     codec_options=self.__write_response_codec_options).copy()\r\n>   File \"/usr/lib/python2.7/site-packages/pymongo/pool.py\", line 244, in command\r\n>     self._raise_connection_failure(error)\r\n>   File \"/usr/lib/python2.7/site-packages/pymongo/pool.py\", line 372, in _raise_connection_failure\r\n>     raise error\r\n> InvalidStringData: strings in documents must be valid UTF-8: '\\x80\\x02N.'\r\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"MongoDB result store backend.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\nfrom datetime import datetime, timedelta\nfrom kombu.utils.objects import cached_property\nfrom kombu.utils.url import maybe_sanitize_url\nfrom kombu.exceptions import EncodeError\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.five import string_t, items\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\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        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, 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\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\nfrom datetime import datetime, timedelta\nfrom kombu.utils.objects import cached_property\nfrom kombu.utils.url import maybe_sanitize_url\nfrom kombu.exceptions import EncodeError\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.five import string_t, items\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\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"}]}