implement new iter_by_chunks() in items

Author: BurnzZ

scrapinghub/client/items.py

@@ -59,3 +59,30 @@ def _modify_iter_params(self, params):
         if offset:
             params['start'] = '{}/{}'.format(self.key, offset)
         return params
+
+    def iter_by_chunks(self, chunksize=10000, *args, **kwargs):
+        """An alternative for reading and processing items by returning a
+        generator of item chunks.
+
+        This is a convenient method for cases when processing a large amount of
+        items from a job isn't ideal in one go due to the large memory needed.
+        Instead, this allows you to process it chunk by chunk.
+
+        You can improve I/O overheads by increasing the chunk value but that
+        would also increase the memory consumption.
+
+        :return: an iterator over a list of elements.
+        :rtype: :class:`collections.Iterable`
+        """
+
+        processed = 0
+        while True:
+            next_key = self.key + '/' + str(processed)
+            items = [
+                item for item in self.iter(
+                    count=chunksize, start=next_key, *args, **kwargs)
+            ]
+            yield items
+            processed += len(items)
+            if len(items) < chunksize:
+                break

tests/client/test_items.py

@@ -36,3 +36,20 @@ def test_items_list(spider, json_and_msgpack):
     assert o[0] == {'id': 0, 'data': 'data0'}
     assert o[1] == {'id': 1, 'data': 'data1'}
     assert o[2] == {'id': 2, 'data': 'data2'}
+
+
+def test_items_iter_by_chunks(spider, json_and_msgpack):
+    job = spider.jobs.run(meta={'state': 'running'})
+    _add_test_items(job)
+
+    o = job.items.iter_by_chunks(2)
+    assert next(o) == [
+        {'id': 0, 'data': 'data0'},
+        {'id': 1, 'data': 'data1'},
+    ]
+    assert next(o) == [
+        {'id': 2, 'data': 'data2'},
+    ]
+    next(o)
+    with pytest.raises(StopIteration):
+        next(o)