Fix list_iter repeats with filter, document iter params

docs/conf.py

@@ -41,6 +41,11 @@
 # ones.
 extensions = ['sphinx_scrapy', 'sphinx_reredirects']
 
+intersphinx_mapping = {
+    'requests': ('https://requests.readthedocs.io/en/latest/', None),
+}
+scrapy_intersphinx_enable = ['zyte']
+
 redirects = {
     "quickstart": "intro/install.html",
     "client/overview": "../use/overview.html",

scrapinghub/client/items.py

@@ -17,6 +17,11 @@ class Items(_DownloadableProxyMixin, _ItemsResourceProxy):
     :meth:`iter` method (all params and available filters are same for
     both methods).
 
+    Besides the parameters of the :ref:`zyte:api-items`, :meth:`iter`
+    supports a *filter* parameter: a list of ``(field, operator, values)``
+    tuples, all of which an item must match, as shown in the last example
+    below.
+
     Usage:
 
     - retrieve all scraped items from a job::
@@ -92,7 +97,7 @@ def _modify_iter_params(self, params):
 
     def list_iter(self, chunksize=1000, *args, **kwargs):
         """An alternative interface for reading items by returning them
-        as a generator which yields lists of items sized as `chunksize`.
+        as a generator which yields lists of up to *chunksize* items.
 
         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.
@@ -101,17 +106,22 @@ def list_iter(self, chunksize=1000, *args, **kwargs):
         You can improve I/O overheads by increasing the chunk value but that
         would also increase the memory consumption.
 
-        :param chunksize: size of list to be returned per iteration
-        :param start: offset to specify the start of the item iteration
-        :param count: overall number of items to be returned, which is broken
-            down by `chunksize`.
+        *start* is the index of the first item to read, and *count* the
+        overall number of items to return. Other parameters, e.g. *filter*,
+        work as in :meth:`iter`.
 
         :return: an iterator over items, yielding lists of items.
         :rtype: :class:`collections.abc.Iterable`
         """
 
         start = kwargs.pop("start", 0)
         count = kwargs.pop("count", sys.maxsize)
+        meta = kwargs.pop("meta", None) or []
+        if isinstance(meta, str):
+            meta = [meta]
+        drop_key = "_key" not in meta
+        if drop_key:
+            meta = [*meta, "_key"]
         processed = 0
 
         while True:
@@ -120,11 +130,18 @@ def list_iter(self, chunksize=1000, *args, **kwargs):
                 chunksize = count - processed
             items = [
                 item for item in self.iter(
-                    count=chunksize, start=next_key, *args, **kwargs)
+                    count=chunksize, start=next_key, meta=meta,
+                    *args, **kwargs)
             ]
+            if items:
+                # Filters may skip items, so resume right after the last
+                # item returned.
+                start = int(items[-1]["_key"].rsplit("/", 1)[1]) + 1
+            if drop_key:
+                for item in items:
+                    del item["_key"]
             yield items
             processed += len(items)
-            start += len(items)
             if processed >= count:
                 break
             if len(items) < chunksize:

scrapinghub/client/proxy.py

@@ -102,9 +102,16 @@ def close(self, block=True):
 class _DownloadableProxyMixin(object):
 
     def iter(self, _path=None, count=None, requests_params=None, **apiparams):
-        """A general method to iterate through elements.
+        """Iterate through elements.
+
+        *count* limits the number of elements returned. It is best to always
+        set it, or to set other pagination parameters, to avoid timeouts.
+
+        *apiparams* are sent to the :ref:`Scrapy Cloud HTTP API
+        <zyte:scrapycloud>` as query parameters, e.g. ``startts`` or
+        ``meta``, and *requests_params* are passed as keyword arguments to
+        :func:`requests.request`, e.g. ``{'timeout': 60}``.
 
-        :param count: limit amount of elements.
         :return: an iterator over elements list.
         :rtype: :class:`collections.abc.Iterable`
         """

tests/client/test_items.py

@@ -1,6 +1,9 @@
+import mock
 import pytest
 from six.moves import range
 
+from scrapinghub.client.items import Items
+
 from .utils import normalize_job_for_tests
 
 
@@ -103,3 +106,31 @@ def test_items_list_iter_with_start_and_count_2(spider, json_and_msgpack):
     ]
     with pytest.raises(StopIteration):
         next(o)
+
+
+@pytest.mark.parametrize('meta, keep_key', [
+    ({}, False),
+    ({'meta': '_key'}, True),
+])
+def test_items_list_iter_with_skipped_items(meta, keep_key):
+    """Items skipped by the server, e.g. due to a filter, must not cause
+    list_iter to return the same items more than once."""
+    matching = [2, 5, 6, 9]
+
+    def iter_(count, start, meta):
+        assert '_key' in meta
+        offset = int(start.rsplit('/', 1)[1])
+        for index in [i for i in matching if i >= offset][:count]:
+            yield {'_key': '1/2/3/{}'.format(index), 'id': index}
+
+    def item(index):
+        if keep_key:
+            return {'_key': '1/2/3/{}'.format(index), 'id': index}
+        return {'id': index}
+
+    items = Items.__new__(Items)
+    items.key = '1/2/3'
+    with mock.patch.object(Items, 'iter', side_effect=iter_):
+        assert list(items.list_iter(chunksize=1, **meta)) == [
+            [item(i)] for i in matching
+        ] + [[]]