diff --git a/docs/conf.py b/docs/conf.py index d3889c1e..97e241cc 100644 --- a/docs/conf.py +++ b/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", diff --git a/scrapinghub/client/items.py b/scrapinghub/client/items.py index e9be38c0..5469fe50 100644 --- a/scrapinghub/client/items.py +++ b/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,10 +106,9 @@ 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` @@ -112,6 +116,12 @@ def list_iter(self, chunksize=1000, *args, **kwargs): 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: diff --git a/scrapinghub/client/proxy.py b/scrapinghub/client/proxy.py index f7d10b4d..9928c8c4 100644 --- a/scrapinghub/client/proxy.py +++ b/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 + ` 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` """ diff --git a/tests/client/test_items.py b/tests/client/test_items.py index 79f68c33..732e55bf 100644 --- a/tests/client/test_items.py +++ b/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 + ] + [[]]