We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[cliquet] Add timeout option for Redis client
original: https://github.com/mozilla-services/cliquet/issues/582

all in the title.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `kinto/core/storage/redis.py`
Content:
```
1 from __future__ import absolute_import, unicode_literals
2 from functools import wraps
3 
4 import redis
5 from six.moves.urllib import parse as urlparse
6 
7 from kinto.core import utils, logger
8 from kinto.core.storage import (
9     exceptions, DEFAULT_ID_FIELD,
10     DEFAULT_MODIFIED_FIELD, DEFAULT_DELETED_FIELD)
11 from kinto.core.storage.memory import MemoryBasedStorage
12 
13 
14 def wrap_redis_error(func):
15     @wraps(func)
16     def wrapped(*args, **kwargs):
17         try:
18             return func(*args, **kwargs)
19         except redis.RedisError as e:
20             logger.exception(e)
21             raise exceptions.BackendError(original=e)
22     return wrapped
23 
24 
25 def create_from_config(config, prefix=''):
26     """Redis client instantiation from settings.
27     """
28     settings = config.get_settings()
29     uri = settings[prefix + 'url']
30     uri = urlparse.urlparse(uri)
31     pool_size = int(settings[prefix + 'pool_size'])
32     kwargs = {
33         "max_connections": pool_size,
34         "host": uri.hostname or 'localhost',
35         "port": uri.port or 6379,
36         "password": uri.password or None,
37         "db": int(uri.path[1:]) if uri.path else 0
38     }
39     connection_pool = redis.BlockingConnectionPool(**kwargs)
40     return redis.StrictRedis(connection_pool=connection_pool)
41 
42 
43 class Storage(MemoryBasedStorage):
44     """Storage backend implementation using Redis.
45 
46     .. warning::
47 
48         Useful for very low server load, but won't scale since records sorting
49         and filtering are performed in memory.
50 
51     Enable in configuration::
52 
53         kinto.storage_backend = kinto.core.storage.redis
54 
55     *(Optional)* Instance location URI can be customized::
56 
57         kinto.storage_url = redis://localhost:6379/0
58 
59     A threaded connection pool is enabled by default::
60 
61         kinto.storage_pool_size = 50
62     """
63 
64     def __init__(self, client, *args, **kwargs):
65         super(Storage, self).__init__(*args, **kwargs)
66         self._client = client
67 
68     @property
69     def settings(self):
70         return dict(self._client.connection_pool.connection_kwargs)
71 
72     def _encode(self, record):
73         return utils.json.dumps(record)
74 
75     def _decode(self, record):
76         return utils.json.loads(record.decode('utf-8'))
77 
78     @wrap_redis_error
79     def flush(self, auth=None):
80         self._client.flushdb()
81 
82     @wrap_redis_error
83     def collection_timestamp(self, collection_id, parent_id, auth=None):
84         timestamp = self._client.get(
85             '{0}.{1}.timestamp'.format(collection_id, parent_id))
86         if timestamp:
87             return int(timestamp)
88         return self._bump_timestamp(collection_id, parent_id)
89 
90     @wrap_redis_error
91     def _bump_timestamp(self, collection_id, parent_id, record=None,
92                         modified_field=None, last_modified=None):
93 
94         key = '{0}.{1}.timestamp'.format(collection_id, parent_id)
95         while 1:
96             with self._client.pipeline() as pipe:
97                 try:
98                     pipe.watch(key)
99                     previous = pipe.get(key)
100                     pipe.multi()
101                     # XXX factorize code from memory and redis backends.
102                     is_specified = (record is not None and
103                                     modified_field in record or
104                                     last_modified is not None)
105                     if is_specified:
106                         # If there is a timestamp in the new record,
107                         # try to use it.
108                         if last_modified is not None:
109                             current = last_modified
110                         else:
111                             current = record[modified_field]
112                     else:
113                         current = utils.msec_time()
114 
115                     if previous and int(previous) >= current:
116                         collection_timestamp = int(previous) + 1
117                     else:
118                         collection_timestamp = current
119 
120                     # Return the newly generated timestamp as the current one
121                     # only if nothing else was specified.
122                     if not is_specified:
123                         current = collection_timestamp
124 
125                     pipe.set(key, collection_timestamp)
126                     pipe.execute()
127                     return current
128                 except redis.WatchError:  # pragma: no cover
129                     # Our timestamp has been modified by someone else, let's
130                     # retry.
131                     # XXX: untested.
132                     continue
133 
134     @wrap_redis_error
135     def create(self, collection_id, parent_id, record, id_generator=None,
136                unique_fields=None, id_field=DEFAULT_ID_FIELD,
137                modified_field=DEFAULT_MODIFIED_FIELD,
138                auth=None):
139         self.check_unicity(collection_id, parent_id, record,
140                            unique_fields=unique_fields, id_field=id_field,
141                            for_creation=True)
142 
143         record = record.copy()
144         id_generator = id_generator or self.id_generator
145         _id = record.setdefault(id_field, id_generator())
146         self.set_record_timestamp(collection_id, parent_id, record,
147                                   modified_field=modified_field)
148 
149         record_key = '{0}.{1}.{2}.records'.format(collection_id,
150                                                   parent_id,
151                                                   _id)
152         with self._client.pipeline() as multi:
153             multi.set(
154                 record_key,
155                 self._encode(record)
156             )
157             multi.sadd(
158                 '{0}.{1}.records'.format(collection_id, parent_id),
159                 _id
160             )
161             multi.srem(
162                 '{0}.{1}.deleted'.format(collection_id, parent_id),
163                 _id
164             )
165             multi.execute()
166 
167         return record
168 
169     @wrap_redis_error
170     def get(self, collection_id, parent_id, object_id,
171             id_field=DEFAULT_ID_FIELD,
172             modified_field=DEFAULT_MODIFIED_FIELD,
173             auth=None):
174         record_key = '{0}.{1}.{2}.records'.format(collection_id,
175                                                   parent_id,
176                                                   object_id)
177         encoded_item = self._client.get(record_key)
178         if encoded_item is None:
179             raise exceptions.RecordNotFoundError(object_id)
180 
181         return self._decode(encoded_item)
182 
183     @wrap_redis_error
184     def update(self, collection_id, parent_id, object_id, record,
185                unique_fields=None, id_field=DEFAULT_ID_FIELD,
186                modified_field=DEFAULT_MODIFIED_FIELD,
187                auth=None):
188         record = record.copy()
189         record[id_field] = object_id
190         self.check_unicity(collection_id, parent_id, record,
191                            unique_fields=unique_fields, id_field=id_field)
192 
193         self.set_record_timestamp(collection_id, parent_id, record,
194                                   modified_field=modified_field)
195 
196         record_key = '{0}.{1}.{2}.records'.format(collection_id,
197                                                   parent_id,
198                                                   object_id)
199         with self._client.pipeline() as multi:
200             multi.set(
201                 record_key,
202                 self._encode(record)
203             )
204             multi.sadd(
205                 '{0}.{1}.records'.format(collection_id, parent_id),
206                 object_id
207             )
208             multi.execute()
209 
210         return record
211 
212     @wrap_redis_error
213     def delete(self, collection_id, parent_id, object_id,
214                id_field=DEFAULT_ID_FIELD, with_deleted=True,
215                modified_field=DEFAULT_MODIFIED_FIELD,
216                deleted_field=DEFAULT_DELETED_FIELD,
217                auth=None, last_modified=None):
218         record_key = '{0}.{1}.{2}.records'.format(collection_id,
219                                                   parent_id,
220                                                   object_id)
221         with self._client.pipeline() as multi:
222             multi.get(record_key)
223             multi.delete(record_key)
224             multi.srem(
225                 '{0}.{1}.records'.format(collection_id, parent_id),
226                 object_id
227             )
228             responses = multi.execute()
229 
230         encoded_item = responses[0]
231         if encoded_item is None:
232             raise exceptions.RecordNotFoundError(object_id)
233 
234         existing = self._decode(encoded_item)
235 
236         # Need to delete the last_modified field.
237         del existing[modified_field]
238 
239         self.set_record_timestamp(collection_id, parent_id, existing,
240                                   modified_field=modified_field,
241                                   last_modified=last_modified)
242         existing = self.strip_deleted_record(collection_id, parent_id,
243                                              existing)
244 
245         if with_deleted:
246             deleted_record_key = '{0}.{1}.{2}.deleted'.format(collection_id,
247                                                               parent_id,
248                                                               object_id)
249             with self._client.pipeline() as multi:
250                 multi.set(
251                     deleted_record_key,
252                     self._encode(existing)
253                 )
254                 multi.sadd(
255                     '{0}.{1}.deleted'.format(collection_id, parent_id),
256                     object_id
257                 )
258                 multi.execute()
259 
260         return existing
261 
262     @wrap_redis_error
263     def purge_deleted(self, collection_id, parent_id, before=None,
264                       id_field=DEFAULT_ID_FIELD,
265                       modified_field=DEFAULT_MODIFIED_FIELD,
266                       auth=None):
267         deleted_ids = '{0}.{1}.deleted'.format(collection_id, parent_id)
268         ids = self._client.smembers(deleted_ids)
269 
270         keys = ['{0}.{1}.{2}.deleted'.format(collection_id, parent_id,
271                                              _id.decode('utf-8'))
272                 for _id in ids]
273 
274         if len(keys) == 0:
275             deleted = []
276         else:
277             encoded_results = self._client.mget(keys)
278             deleted = [self._decode(r) for r in encoded_results if r]
279         if before is not None:
280             to_remove = [d['id'] for d in deleted
281                          if d[modified_field] < before]
282         else:
283             to_remove = [d['id'] for d in deleted]
284 
285         if len(to_remove) > 0:
286             with self._client.pipeline() as pipe:
287                 pipe.delete(*['{0}.{1}.{2}.deleted'.format(
288                     collection_id, parent_id, _id) for _id in to_remove])
289                 pipe.srem(deleted_ids, *to_remove)
290                 pipe.execute()
291         number_deleted = len(to_remove)
292         return number_deleted
293 
294     @wrap_redis_error
295     def get_all(self, collection_id, parent_id, filters=None, sorting=None,
296                 pagination_rules=None, limit=None, include_deleted=False,
297                 id_field=DEFAULT_ID_FIELD,
298                 modified_field=DEFAULT_MODIFIED_FIELD,
299                 deleted_field=DEFAULT_DELETED_FIELD,
300                 auth=None):
301         records_ids_key = '{0}.{1}.records'.format(collection_id, parent_id)
302         ids = self._client.smembers(records_ids_key)
303 
304         keys = ('{0}.{1}.{2}.records'.format(collection_id, parent_id,
305                                              _id.decode('utf-8'))
306                 for _id in ids)
307 
308         if len(ids) == 0:
309             records = []
310         else:
311             encoded_results = self._client.mget(keys)
312             records = [self._decode(r) for r in encoded_results if r]
313 
314         deleted = []
315         if include_deleted:
316             deleted_ids = '{0}.{1}.deleted'.format(collection_id, parent_id)
317             ids = self._client.smembers(deleted_ids)
318 
319             keys = ['{0}.{1}.{2}.deleted'.format(collection_id, parent_id,
320                                                  _id.decode('utf-8'))
321                     for _id in ids]
322 
323             if len(keys) == 0:
324                 deleted = []
325             else:
326                 encoded_results = self._client.mget(keys)
327                 deleted = [self._decode(r) for r in encoded_results if r]
328 
329         records, count = self.extract_record_set(collection_id,
330                                                  records + deleted,
331                                                  filters, sorting,
332                                                  id_field, deleted_field,
333                                                  pagination_rules, limit)
334 
335         return records, count
336 
337 
338 def load_from_config(config):
339     client = create_from_config(config, prefix='storage_')
340     return Storage(client)
341
```
--- 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/kinto/core/storage/redis.py b/kinto/core/storage/redis.py
--- a/kinto/core/storage/redis.py
+++ b/kinto/core/storage/redis.py
@@ -36,6 +36,10 @@
         "password": uri.password or None,
         "db": int(uri.path[1:]) if uri.path else 0
     }
+    block_timeout = settings.get(prefix + 'pool_timeout')
+    if block_timeout is not None:
+        kwargs["timeout"] = float(block_timeout)
+
     connection_pool = redis.BlockingConnectionPool(**kwargs)
     return redis.StrictRedis(connection_pool=connection_pool)

{"golden_diff": "diff --git a/kinto/core/storage/redis.py b/kinto/core/storage/redis.py\n--- a/kinto/core/storage/redis.py\n+++ b/kinto/core/storage/redis.py\n@@ -36,6 +36,10 @@\n         \"password\": uri.password or None,\n         \"db\": int(uri.path[1:]) if uri.path else 0\n     }\n+    block_timeout = settings.get(prefix + 'pool_timeout')\n+    if block_timeout is not None:\n+        kwargs[\"timeout\"] = float(block_timeout)\n+\n     connection_pool = redis.BlockingConnectionPool(**kwargs)\n     return redis.StrictRedis(connection_pool=connection_pool)\n", "issue": "[cliquet] Add timeout option for Redis client\noriginal: https://github.com/mozilla-services/cliquet/issues/582\n\nall in the title.\n\n", "before_files": [{"content": "from __future__ import absolute_import, unicode_literals\nfrom functools import wraps\n\nimport redis\nfrom six.moves.urllib import parse as urlparse\n\nfrom kinto.core import utils, logger\nfrom kinto.core.storage import (\n    exceptions, DEFAULT_ID_FIELD,\n    DEFAULT_MODIFIED_FIELD, DEFAULT_DELETED_FIELD)\nfrom kinto.core.storage.memory import MemoryBasedStorage\n\n\ndef wrap_redis_error(func):\n    @wraps(func)\n    def wrapped(*args, **kwargs):\n        try:\n            return func(*args, **kwargs)\n        except redis.RedisError as e:\n            logger.exception(e)\n            raise exceptions.BackendError(original=e)\n    return wrapped\n\n\ndef create_from_config(config, prefix=''):\n    \"\"\"Redis client instantiation from settings.\n    \"\"\"\n    settings = config.get_settings()\n    uri = settings[prefix + 'url']\n    uri = urlparse.urlparse(uri)\n    pool_size = int(settings[prefix + 'pool_size'])\n    kwargs = {\n        \"max_connections\": pool_size,\n        \"host\": uri.hostname or 'localhost',\n        \"port\": uri.port or 6379,\n        \"password\": uri.password or None,\n        \"db\": int(uri.path[1:]) if uri.path else 0\n    }\n    connection_pool = redis.BlockingConnectionPool(**kwargs)\n    return redis.StrictRedis(connection_pool=connection_pool)\n\n\nclass Storage(MemoryBasedStorage):\n    \"\"\"Storage backend implementation using Redis.\n\n    .. warning::\n\n        Useful for very low server load, but won't scale since records sorting\n        and filtering are performed in memory.\n\n    Enable in configuration::\n\n        kinto.storage_backend = kinto.core.storage.redis\n\n    *(Optional)* Instance location URI can be customized::\n\n        kinto.storage_url = redis://localhost:6379/0\n\n    A threaded connection pool is enabled by default::\n\n        kinto.storage_pool_size = 50\n    \"\"\"\n\n    def __init__(self, client, *args, **kwargs):\n        super(Storage, self).__init__(*args, **kwargs)\n        self._client = client\n\n    @property\n    def settings(self):\n        return dict(self._client.connection_pool.connection_kwargs)\n\n    def _encode(self, record):\n        return utils.json.dumps(record)\n\n    def _decode(self, record):\n        return utils.json.loads(record.decode('utf-8'))\n\n    @wrap_redis_error\n    def flush(self, auth=None):\n        self._client.flushdb()\n\n    @wrap_redis_error\n    def collection_timestamp(self, collection_id, parent_id, auth=None):\n        timestamp = self._client.get(\n            '{0}.{1}.timestamp'.format(collection_id, parent_id))\n        if timestamp:\n            return int(timestamp)\n        return self._bump_timestamp(collection_id, parent_id)\n\n    @wrap_redis_error\n    def _bump_timestamp(self, collection_id, parent_id, record=None,\n                        modified_field=None, last_modified=None):\n\n        key = '{0}.{1}.timestamp'.format(collection_id, parent_id)\n        while 1:\n            with self._client.pipeline() as pipe:\n                try:\n                    pipe.watch(key)\n                    previous = pipe.get(key)\n                    pipe.multi()\n                    # XXX factorize code from memory and redis backends.\n                    is_specified = (record is not None and\n                                    modified_field in record or\n                                    last_modified is not None)\n                    if is_specified:\n                        # If there is a timestamp in the new record,\n                        # try to use it.\n                        if last_modified is not None:\n                            current = last_modified\n                        else:\n                            current = record[modified_field]\n                    else:\n                        current = utils.msec_time()\n\n                    if previous and int(previous) >= current:\n                        collection_timestamp = int(previous) + 1\n                    else:\n                        collection_timestamp = current\n\n                    # Return the newly generated timestamp as the current one\n                    # only if nothing else was specified.\n                    if not is_specified:\n                        current = collection_timestamp\n\n                    pipe.set(key, collection_timestamp)\n                    pipe.execute()\n                    return current\n                except redis.WatchError:  # pragma: no cover\n                    # Our timestamp has been modified by someone else, let's\n                    # retry.\n                    # XXX: untested.\n                    continue\n\n    @wrap_redis_error\n    def create(self, collection_id, parent_id, record, id_generator=None,\n               unique_fields=None, id_field=DEFAULT_ID_FIELD,\n               modified_field=DEFAULT_MODIFIED_FIELD,\n               auth=None):\n        self.check_unicity(collection_id, parent_id, record,\n                           unique_fields=unique_fields, id_field=id_field,\n                           for_creation=True)\n\n        record = record.copy()\n        id_generator = id_generator or self.id_generator\n        _id = record.setdefault(id_field, id_generator())\n        self.set_record_timestamp(collection_id, parent_id, record,\n                                  modified_field=modified_field)\n\n        record_key = '{0}.{1}.{2}.records'.format(collection_id,\n                                                  parent_id,\n                                                  _id)\n        with self._client.pipeline() as multi:\n            multi.set(\n                record_key,\n                self._encode(record)\n            )\n            multi.sadd(\n                '{0}.{1}.records'.format(collection_id, parent_id),\n                _id\n            )\n            multi.srem(\n                '{0}.{1}.deleted'.format(collection_id, parent_id),\n                _id\n            )\n            multi.execute()\n\n        return record\n\n    @wrap_redis_error\n    def get(self, collection_id, parent_id, object_id,\n            id_field=DEFAULT_ID_FIELD,\n            modified_field=DEFAULT_MODIFIED_FIELD,\n            auth=None):\n        record_key = '{0}.{1}.{2}.records'.format(collection_id,\n                                                  parent_id,\n                                                  object_id)\n        encoded_item = self._client.get(record_key)\n        if encoded_item is None:\n            raise exceptions.RecordNotFoundError(object_id)\n\n        return self._decode(encoded_item)\n\n    @wrap_redis_error\n    def update(self, collection_id, parent_id, object_id, record,\n               unique_fields=None, id_field=DEFAULT_ID_FIELD,\n               modified_field=DEFAULT_MODIFIED_FIELD,\n               auth=None):\n        record = record.copy()\n        record[id_field] = object_id\n        self.check_unicity(collection_id, parent_id, record,\n                           unique_fields=unique_fields, id_field=id_field)\n\n        self.set_record_timestamp(collection_id, parent_id, record,\n                                  modified_field=modified_field)\n\n        record_key = '{0}.{1}.{2}.records'.format(collection_id,\n                                                  parent_id,\n                                                  object_id)\n        with self._client.pipeline() as multi:\n            multi.set(\n                record_key,\n                self._encode(record)\n            )\n            multi.sadd(\n                '{0}.{1}.records'.format(collection_id, parent_id),\n                object_id\n            )\n            multi.execute()\n\n        return record\n\n    @wrap_redis_error\n    def delete(self, collection_id, parent_id, object_id,\n               id_field=DEFAULT_ID_FIELD, with_deleted=True,\n               modified_field=DEFAULT_MODIFIED_FIELD,\n               deleted_field=DEFAULT_DELETED_FIELD,\n               auth=None, last_modified=None):\n        record_key = '{0}.{1}.{2}.records'.format(collection_id,\n                                                  parent_id,\n                                                  object_id)\n        with self._client.pipeline() as multi:\n            multi.get(record_key)\n            multi.delete(record_key)\n            multi.srem(\n                '{0}.{1}.records'.format(collection_id, parent_id),\n                object_id\n            )\n            responses = multi.execute()\n\n        encoded_item = responses[0]\n        if encoded_item is None:\n            raise exceptions.RecordNotFoundError(object_id)\n\n        existing = self._decode(encoded_item)\n\n        # Need to delete the last_modified field.\n        del existing[modified_field]\n\n        self.set_record_timestamp(collection_id, parent_id, existing,\n                                  modified_field=modified_field,\n                                  last_modified=last_modified)\n        existing = self.strip_deleted_record(collection_id, parent_id,\n                                             existing)\n\n        if with_deleted:\n            deleted_record_key = '{0}.{1}.{2}.deleted'.format(collection_id,\n                                                              parent_id,\n                                                              object_id)\n            with self._client.pipeline() as multi:\n                multi.set(\n                    deleted_record_key,\n                    self._encode(existing)\n                )\n                multi.sadd(\n                    '{0}.{1}.deleted'.format(collection_id, parent_id),\n                    object_id\n                )\n                multi.execute()\n\n        return existing\n\n    @wrap_redis_error\n    def purge_deleted(self, collection_id, parent_id, before=None,\n                      id_field=DEFAULT_ID_FIELD,\n                      modified_field=DEFAULT_MODIFIED_FIELD,\n                      auth=None):\n        deleted_ids = '{0}.{1}.deleted'.format(collection_id, parent_id)\n        ids = self._client.smembers(deleted_ids)\n\n        keys = ['{0}.{1}.{2}.deleted'.format(collection_id, parent_id,\n                                             _id.decode('utf-8'))\n                for _id in ids]\n\n        if len(keys) == 0:\n            deleted = []\n        else:\n            encoded_results = self._client.mget(keys)\n            deleted = [self._decode(r) for r in encoded_results if r]\n        if before is not None:\n            to_remove = [d['id'] for d in deleted\n                         if d[modified_field] < before]\n        else:\n            to_remove = [d['id'] for d in deleted]\n\n        if len(to_remove) > 0:\n            with self._client.pipeline() as pipe:\n                pipe.delete(*['{0}.{1}.{2}.deleted'.format(\n                    collection_id, parent_id, _id) for _id in to_remove])\n                pipe.srem(deleted_ids, *to_remove)\n                pipe.execute()\n        number_deleted = len(to_remove)\n        return number_deleted\n\n    @wrap_redis_error\n    def get_all(self, collection_id, parent_id, filters=None, sorting=None,\n                pagination_rules=None, limit=None, include_deleted=False,\n                id_field=DEFAULT_ID_FIELD,\n                modified_field=DEFAULT_MODIFIED_FIELD,\n                deleted_field=DEFAULT_DELETED_FIELD,\n                auth=None):\n        records_ids_key = '{0}.{1}.records'.format(collection_id, parent_id)\n        ids = self._client.smembers(records_ids_key)\n\n        keys = ('{0}.{1}.{2}.records'.format(collection_id, parent_id,\n                                             _id.decode('utf-8'))\n                for _id in ids)\n\n        if len(ids) == 0:\n            records = []\n        else:\n            encoded_results = self._client.mget(keys)\n            records = [self._decode(r) for r in encoded_results if r]\n\n        deleted = []\n        if include_deleted:\n            deleted_ids = '{0}.{1}.deleted'.format(collection_id, parent_id)\n            ids = self._client.smembers(deleted_ids)\n\n            keys = ['{0}.{1}.{2}.deleted'.format(collection_id, parent_id,\n                                                 _id.decode('utf-8'))\n                    for _id in ids]\n\n            if len(keys) == 0:\n                deleted = []\n            else:\n                encoded_results = self._client.mget(keys)\n                deleted = [self._decode(r) for r in encoded_results if r]\n\n        records, count = self.extract_record_set(collection_id,\n                                                 records + deleted,\n                                                 filters, sorting,\n                                                 id_field, deleted_field,\n                                                 pagination_rules, limit)\n\n        return records, count\n\n\ndef load_from_config(config):\n    client = create_from_config(config, prefix='storage_')\n    return Storage(client)\n", "path": "kinto/core/storage/redis.py"}], "after_files": [{"content": "from __future__ import absolute_import, unicode_literals\nfrom functools import wraps\n\nimport redis\nfrom six.moves.urllib import parse as urlparse\n\nfrom kinto.core import utils, logger\nfrom kinto.core.storage import (\n    exceptions, DEFAULT_ID_FIELD,\n    DEFAULT_MODIFIED_FIELD, DEFAULT_DELETED_FIELD)\nfrom kinto.core.storage.memory import MemoryBasedStorage\n\n\ndef wrap_redis_error(func):\n    @wraps(func)\n    def wrapped(*args, **kwargs):\n        try:\n            return func(*args, **kwargs)\n        except redis.RedisError as e:\n            logger.exception(e)\n            raise exceptions.BackendError(original=e)\n    return wrapped\n\n\ndef create_from_config(config, prefix=''):\n    \"\"\"Redis client instantiation from settings.\n    \"\"\"\n    settings = config.get_settings()\n    uri = settings[prefix + 'url']\n    uri = urlparse.urlparse(uri)\n    pool_size = int(settings[prefix + 'pool_size'])\n    kwargs = {\n        \"max_connections\": pool_size,\n        \"host\": uri.hostname or 'localhost',\n        \"port\": uri.port or 6379,\n        \"password\": uri.password or None,\n        \"db\": int(uri.path[1:]) if uri.path else 0\n    }\n    block_timeout = settings.get(prefix + 'pool_timeout')\n    if block_timeout is not None:\n        kwargs[\"timeout\"] = float(block_timeout)\n\n    connection_pool = redis.BlockingConnectionPool(**kwargs)\n    return redis.StrictRedis(connection_pool=connection_pool)\n\n\nclass Storage(MemoryBasedStorage):\n    \"\"\"Storage backend implementation using Redis.\n\n    .. warning::\n\n        Useful for very low server load, but won't scale since records sorting\n        and filtering are performed in memory.\n\n    Enable in configuration::\n\n        kinto.storage_backend = kinto.core.storage.redis\n\n    *(Optional)* Instance location URI can be customized::\n\n        kinto.storage_url = redis://localhost:6379/0\n\n    A threaded connection pool is enabled by default::\n\n        kinto.storage_pool_size = 50\n    \"\"\"\n\n    def __init__(self, client, *args, **kwargs):\n        super(Storage, self).__init__(*args, **kwargs)\n        self._client = client\n\n    @property\n    def settings(self):\n        return dict(self._client.connection_pool.connection_kwargs)\n\n    def _encode(self, record):\n        return utils.json.dumps(record)\n\n    def _decode(self, record):\n        return utils.json.loads(record.decode('utf-8'))\n\n    @wrap_redis_error\n    def flush(self, auth=None):\n        self._client.flushdb()\n\n    @wrap_redis_error\n    def collection_timestamp(self, collection_id, parent_id, auth=None):\n        timestamp = self._client.get(\n            '{0}.{1}.timestamp'.format(collection_id, parent_id))\n        if timestamp:\n            return int(timestamp)\n        return self._bump_timestamp(collection_id, parent_id)\n\n    @wrap_redis_error\n    def _bump_timestamp(self, collection_id, parent_id, record=None,\n                        modified_field=None, last_modified=None):\n\n        key = '{0}.{1}.timestamp'.format(collection_id, parent_id)\n        while 1:\n            with self._client.pipeline() as pipe:\n                try:\n                    pipe.watch(key)\n                    previous = pipe.get(key)\n                    pipe.multi()\n                    # XXX factorize code from memory and redis backends.\n                    is_specified = (record is not None and\n                                    modified_field in record or\n                                    last_modified is not None)\n                    if is_specified:\n                        # If there is a timestamp in the new record,\n                        # try to use it.\n                        if last_modified is not None:\n                            current = last_modified\n                        else:\n                            current = record[modified_field]\n                    else:\n                        current = utils.msec_time()\n\n                    if previous and int(previous) >= current:\n                        collection_timestamp = int(previous) + 1\n                    else:\n                        collection_timestamp = current\n\n                    # Return the newly generated timestamp as the current one\n                    # only if nothing else was specified.\n                    if not is_specified:\n                        current = collection_timestamp\n\n                    pipe.set(key, collection_timestamp)\n                    pipe.execute()\n                    return current\n                except redis.WatchError:  # pragma: no cover\n                    # Our timestamp has been modified by someone else, let's\n                    # retry.\n                    # XXX: untested.\n                    continue\n\n    @wrap_redis_error\n    def create(self, collection_id, parent_id, record, id_generator=None,\n               unique_fields=None, id_field=DEFAULT_ID_FIELD,\n               modified_field=DEFAULT_MODIFIED_FIELD,\n               auth=None):\n        self.check_unicity(collection_id, parent_id, record,\n                           unique_fields=unique_fields, id_field=id_field,\n                           for_creation=True)\n\n        record = record.copy()\n        id_generator = id_generator or self.id_generator\n        _id = record.setdefault(id_field, id_generator())\n        self.set_record_timestamp(collection_id, parent_id, record,\n                                  modified_field=modified_field)\n\n        record_key = '{0}.{1}.{2}.records'.format(collection_id,\n                                                  parent_id,\n                                                  _id)\n        with self._client.pipeline() as multi:\n            multi.set(\n                record_key,\n                self._encode(record)\n            )\n            multi.sadd(\n                '{0}.{1}.records'.format(collection_id, parent_id),\n                _id\n            )\n            multi.srem(\n                '{0}.{1}.deleted'.format(collection_id, parent_id),\n                _id\n            )\n            multi.execute()\n\n        return record\n\n    @wrap_redis_error\n    def get(self, collection_id, parent_id, object_id,\n            id_field=DEFAULT_ID_FIELD,\n            modified_field=DEFAULT_MODIFIED_FIELD,\n            auth=None):\n        record_key = '{0}.{1}.{2}.records'.format(collection_id,\n                                                  parent_id,\n                                                  object_id)\n        encoded_item = self._client.get(record_key)\n        if encoded_item is None:\n            raise exceptions.RecordNotFoundError(object_id)\n\n        return self._decode(encoded_item)\n\n    @wrap_redis_error\n    def update(self, collection_id, parent_id, object_id, record,\n               unique_fields=None, id_field=DEFAULT_ID_FIELD,\n               modified_field=DEFAULT_MODIFIED_FIELD,\n               auth=None):\n        record = record.copy()\n        record[id_field] = object_id\n        self.check_unicity(collection_id, parent_id, record,\n                           unique_fields=unique_fields, id_field=id_field)\n\n        self.set_record_timestamp(collection_id, parent_id, record,\n                                  modified_field=modified_field)\n\n        record_key = '{0}.{1}.{2}.records'.format(collection_id,\n                                                  parent_id,\n                                                  object_id)\n        with self._client.pipeline() as multi:\n            multi.set(\n                record_key,\n                self._encode(record)\n            )\n            multi.sadd(\n                '{0}.{1}.records'.format(collection_id, parent_id),\n                object_id\n            )\n            multi.execute()\n\n        return record\n\n    @wrap_redis_error\n    def delete(self, collection_id, parent_id, object_id,\n               id_field=DEFAULT_ID_FIELD, with_deleted=True,\n               modified_field=DEFAULT_MODIFIED_FIELD,\n               deleted_field=DEFAULT_DELETED_FIELD,\n               auth=None, last_modified=None):\n        record_key = '{0}.{1}.{2}.records'.format(collection_id,\n                                                  parent_id,\n                                                  object_id)\n        with self._client.pipeline() as multi:\n            multi.get(record_key)\n            multi.delete(record_key)\n            multi.srem(\n                '{0}.{1}.records'.format(collection_id, parent_id),\n                object_id\n            )\n            responses = multi.execute()\n\n        encoded_item = responses[0]\n        if encoded_item is None:\n            raise exceptions.RecordNotFoundError(object_id)\n\n        existing = self._decode(encoded_item)\n\n        # Need to delete the last_modified field.\n        del existing[modified_field]\n\n        self.set_record_timestamp(collection_id, parent_id, existing,\n                                  modified_field=modified_field,\n                                  last_modified=last_modified)\n        existing = self.strip_deleted_record(collection_id, parent_id,\n                                             existing)\n\n        if with_deleted:\n            deleted_record_key = '{0}.{1}.{2}.deleted'.format(collection_id,\n                                                              parent_id,\n                                                              object_id)\n            with self._client.pipeline() as multi:\n                multi.set(\n                    deleted_record_key,\n                    self._encode(existing)\n                )\n                multi.sadd(\n                    '{0}.{1}.deleted'.format(collection_id, parent_id),\n                    object_id\n                )\n                multi.execute()\n\n        return existing\n\n    @wrap_redis_error\n    def purge_deleted(self, collection_id, parent_id, before=None,\n                      id_field=DEFAULT_ID_FIELD,\n                      modified_field=DEFAULT_MODIFIED_FIELD,\n                      auth=None):\n        deleted_ids = '{0}.{1}.deleted'.format(collection_id, parent_id)\n        ids = self._client.smembers(deleted_ids)\n\n        keys = ['{0}.{1}.{2}.deleted'.format(collection_id, parent_id,\n                                             _id.decode('utf-8'))\n                for _id in ids]\n\n        if len(keys) == 0:\n            deleted = []\n        else:\n            encoded_results = self._client.mget(keys)\n            deleted = [self._decode(r) for r in encoded_results if r]\n        if before is not None:\n            to_remove = [d['id'] for d in deleted\n                         if d[modified_field] < before]\n        else:\n            to_remove = [d['id'] for d in deleted]\n\n        if len(to_remove) > 0:\n            with self._client.pipeline() as pipe:\n                pipe.delete(*['{0}.{1}.{2}.deleted'.format(\n                    collection_id, parent_id, _id) for _id in to_remove])\n                pipe.srem(deleted_ids, *to_remove)\n                pipe.execute()\n        number_deleted = len(to_remove)\n        return number_deleted\n\n    @wrap_redis_error\n    def get_all(self, collection_id, parent_id, filters=None, sorting=None,\n                pagination_rules=None, limit=None, include_deleted=False,\n                id_field=DEFAULT_ID_FIELD,\n                modified_field=DEFAULT_MODIFIED_FIELD,\n                deleted_field=DEFAULT_DELETED_FIELD,\n                auth=None):\n        records_ids_key = '{0}.{1}.records'.format(collection_id, parent_id)\n        ids = self._client.smembers(records_ids_key)\n\n        keys = ('{0}.{1}.{2}.records'.format(collection_id, parent_id,\n                                             _id.decode('utf-8'))\n                for _id in ids)\n\n        if len(ids) == 0:\n            records = []\n        else:\n            encoded_results = self._client.mget(keys)\n            records = [self._decode(r) for r in encoded_results if r]\n\n        deleted = []\n        if include_deleted:\n            deleted_ids = '{0}.{1}.deleted'.format(collection_id, parent_id)\n            ids = self._client.smembers(deleted_ids)\n\n            keys = ['{0}.{1}.{2}.deleted'.format(collection_id, parent_id,\n                                                 _id.decode('utf-8'))\n                    for _id in ids]\n\n            if len(keys) == 0:\n                deleted = []\n            else:\n                encoded_results = self._client.mget(keys)\n                deleted = [self._decode(r) for r in encoded_results if r]\n\n        records, count = self.extract_record_set(collection_id,\n                                                 records + deleted,\n                                                 filters, sorting,\n                                                 id_field, deleted_field,\n                                                 pagination_rules, limit)\n\n        return records, count\n\n\ndef load_from_config(config):\n    client = create_from_config(config, prefix='storage_')\n    return Storage(client)\n", "path": "kinto/core/storage/redis.py"}]}