fix: Keep pagination iterators advancing past fully-filtered pages

Author: vdusek

src/apify_client/_pagination.py

@@ -41,8 +41,11 @@ def get_items_iterator(
     is used for offset bookkeeping (the Apify API's `count` reflects items scanned, which can exceed items returned when
     filters are applied).
 
-    Iteration stops when a page returns no items or when the user-requested `limit` is reached. The `total` field is
-    intentionally not consulted, because it can change between calls.
+    Iteration stops when a page scans no items (`count` is `0`, or `items` is empty when `count` is absent) or when the
+    user-requested `limit` is reached. A page can scan items while returning none — filters like `clean` drop items from
+    `items` but still count toward `count` — so terminating on scanned rather than returned items keeps the iterator
+    advancing across fully-filtered pages. The `total` field is intentionally not consulted, because it can change
+    between calls.
 
     Args:
         callback: Function returning a single page of items.
@@ -62,9 +65,10 @@ def get_items_iterator(
         )
         yield from current_page.items
 
-        fetched_items += max(getattr(current_page, 'count', 0), len(current_page.items))
+        page_scanned = max(getattr(current_page, 'count', 0), len(current_page.items))
+        fetched_items += page_scanned
 
-        if not current_page.items or (initial_limit and fetched_items >= initial_limit):
+        if not page_scanned or (initial_limit and fetched_items >= initial_limit):
             break
 
 
@@ -92,9 +96,10 @@ async def get_items_iterator_async(
         for item in current_page.items:
             yield item
 
-        fetched_items += max(getattr(current_page, 'count', 0), len(current_page.items))
+        page_scanned = max(getattr(current_page, 'count', 0), len(current_page.items))
+        fetched_items += page_scanned
 
-        if not current_page.items or (initial_limit and fetched_items >= initial_limit):
+        if not page_scanned or (initial_limit and fetched_items >= initial_limit):
             break
 
 
@@ -124,8 +129,8 @@ def get_cursor_iterator(
     """Yield individual items from cursor-paginated API responses.
 
     Cursor pagination is restricted to the two API responses that expose it: `ListOfKeys` (for key-value store keys) and
-    `ListOfRequests` (for request queue requests). Iteration ends when a page returns no items, the next cursor is
-    `None`, or the user-requested `limit` is reached.
+    `ListOfRequests` (for request queue requests). Iteration ends when a page scans no items, the next cursor is `None`,
+    or the user-requested `limit` is reached.
 
     Args:
         callback: Function returning a single page of items. Receives `cursor` and `limit` kwargs.
@@ -144,12 +149,13 @@ def get_cursor_iterator(
         )
         yield from current_page.items
 
-        fetched_items += max(getattr(current_page, 'count', 0), len(current_page.items))
+        page_scanned = max(getattr(current_page, 'count', 0), len(current_page.items))
+        fetched_items += page_scanned
         cursor = (
             current_page.next_exclusive_start_key if isinstance(current_page, ListOfKeys) else current_page.next_cursor
         )
 
-        if not current_page.items or cursor is None or (initial_limit and fetched_items >= initial_limit):
+        if not page_scanned or cursor is None or (initial_limit and fetched_items >= initial_limit):
             break
 
 
@@ -189,12 +195,13 @@ async def get_cursor_iterator_async(
         for item in current_page.items:
             yield item
 
-        fetched_items += max(getattr(current_page, 'count', 0), len(current_page.items))
+        page_scanned = max(getattr(current_page, 'count', 0), len(current_page.items))
+        fetched_items += page_scanned
         cursor = (
             current_page.next_exclusive_start_key if isinstance(current_page, ListOfKeys) else current_page.next_cursor
         )
 
-        if not current_page.items or cursor is None or (initial_limit and fetched_items >= initial_limit):
+        if not page_scanned or cursor is None or (initial_limit and fetched_items >= initial_limit):
             break
 
 

tests/unit/test_client_pagination.py

@@ -11,6 +11,12 @@
 
 from apify_client import ApifyClient, ApifyClientAsync
 from apify_client import _models as _models_module
+from apify_client._pagination import (
+    get_cursor_iterator,
+    get_cursor_iterator_async,
+    get_items_iterator,
+    get_items_iterator_async,
+)
 from apify_client._resource_clients import (
     ActorCollectionClient,
     ActorCollectionClientAsync,
@@ -617,3 +623,73 @@ async def test_rq_list_requests_iterable_async(
     client: RequestQueueClientAsync = _CLIENT_FACTORIES[client_name](_make_async_client(pagination_server))
     returned_items = [dict(item) async for item in client.iterate_requests(**inputs)]
     assert returned_items == expected_items
+
+
+class _FakeOffsetPage:
+    """Offset-paginated page whose `count` (items scanned) may exceed `len(items)` when filters drop items."""
+
+    def __init__(self, items: list[dict[str, int]], count: int) -> None:
+        self.items = items
+        self.count = count
+
+
+class _FakeCursorPage:
+    """Cursor-paginated page whose `count` (items scanned) may exceed `len(items)` when filters drop items."""
+
+    def __init__(self, items: list[dict[str, int]], count: int, next_cursor: str | None) -> None:
+        self.items = items
+        self.count = count
+        self.next_cursor = next_cursor
+
+
+def test_items_iterator_continues_past_fully_filtered_page() -> None:
+    """A fully-filtered page (`items=[]`, `count>0`) must not stop the offset iterator while more data was scanned."""
+    pages = {
+        0: _FakeOffsetPage(items=[], count=1000),
+        1000: _FakeOffsetPage(items=[{'id': 1}, {'id': 2}], count=2),
+    }
+
+    def _callback(*, limit: int | None = None, offset: int | None = None) -> _FakeOffsetPage:  # noqa: ARG001
+        return pages.get(offset or 0, _FakeOffsetPage(items=[], count=0))
+
+    assert list(get_items_iterator(_callback, chunk_size=1000)) == [{'id': 1}, {'id': 2}]
+
+
+async def test_items_iterator_async_continues_past_fully_filtered_page() -> None:
+    """A fully-filtered page (`items=[]`, `count>0`) must not stop the async offset iterator while more was scanned."""
+    pages = {
+        0: _FakeOffsetPage(items=[], count=1000),
+        1000: _FakeOffsetPage(items=[{'id': 1}, {'id': 2}], count=2),
+    }
+
+    async def _callback(*, limit: int | None = None, offset: int | None = None) -> _FakeOffsetPage:  # noqa: ARG001
+        return pages.get(offset or 0, _FakeOffsetPage(items=[], count=0))
+
+    assert [item async for item in get_items_iterator_async(_callback, chunk_size=1000)] == [{'id': 1}, {'id': 2}]
+
+
+def test_cursor_iterator_continues_past_fully_filtered_page() -> None:
+    """A fully-filtered page (`items=[]`, `count>0`) with a live cursor must not stop the cursor iterator."""
+    pages = {
+        None: _FakeCursorPage(items=[], count=1000, next_cursor='c1'),
+        'c1': _FakeCursorPage(items=[{'id': 1}, {'id': 2}], count=2, next_cursor=None),
+    }
+
+    def _callback(*, limit: int | None = None, cursor: str | None = None) -> _FakeCursorPage:  # noqa: ARG001
+        return pages[cursor]
+
+    assert list(get_cursor_iterator(_callback, chunk_size=1000)) == [{'id': 1}, {'id': 2}]  # ty: ignore[no-matching-overload]
+
+
+async def test_cursor_iterator_async_continues_past_fully_filtered_page() -> None:
+    """A fully-filtered page (`items=[]`, `count>0`) with a live cursor must not stop the async cursor iterator."""
+    pages = {
+        None: _FakeCursorPage(items=[], count=1000, next_cursor='c1'),
+        'c1': _FakeCursorPage(items=[{'id': 1}, {'id': 2}], count=2, next_cursor=None),
+    }
+
+    async def _callback(*, limit: int | None = None, cursor: str | None = None) -> _FakeCursorPage:  # noqa: ARG001
+        return pages[cursor]
+
+    collected = [item async for item in get_cursor_iterator_async(_callback, chunk_size=1000)]  # ty: ignore[no-matching-overload]
+    assert collected == [{'id': 1}, {'id': 2}]