fix: Keep pagination iterators advancing past fully-filtered pages

src/apify_client/_pagination.py

@@ -19,7 +19,7 @@
 class HasItems(Protocol[T]):
     """Structural contract for a single page of results from a paginated API endpoint.
 
-    Implementations must expose `items`. They may optionally expose `count` — the number of items scanned by the API for
+    Implementations must expose `items`. They may optionally expose `count` - the number of items scanned by the API for
     this page, which can exceed `len(items)` when filters drop items from the response. The iterator helpers consult
     `count` opportunistically via `getattr` for offset bookkeeping and fall back to `len(items)` when it is absent.
     """
@@ -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,11 @@ 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 the next cursor is `None` or the user-requested
+    `limit` is reached. Emptiness alone does not stop iteration: server-side filters (such as the request-queue state
+    `filter`) can drop every item on a page while a live cursor still points at more data, so termination relies on the
+    cursor, not on whether a page returned items. Unlike offset responses, cursor responses expose no scanned-item
+    `count`, so `count` cannot be used to detect a fully-filtered page here.
 
     Args:
         callback: Function returning a single page of items. Receives `cursor` and `limit` kwargs.
@@ -144,12 +152,12 @@ def get_cursor_iterator(
         )
         yield from current_page.items
 
-        fetched_items += max(getattr(current_page, 'count', 0), len(current_page.items))
+        fetched_items += len(current_page.items)
         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 cursor is None or (initial_limit and fetched_items >= initial_limit):
             break
 
 
@@ -189,12 +197,12 @@ 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))
+        fetched_items += len(current_page.items)
         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 cursor is None or (initial_limit and fetched_items >= initial_limit):
             break
 
 

tests/unit/test_client_pagination.py

@@ -11,6 +11,13 @@
 
 from apify_client import ApifyClient, ApifyClientAsync
 from apify_client import _models as _models_module
+from apify_client._models import ListOfRequests
+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,
@@ -113,7 +120,7 @@
 )
 
 # Outer wrappers that embed a relaxed list model via `.data`. Their compiled schema pins the inner's schema at
-# construction time, so they need a forced rebuild to pick up the relaxation. The wrappers themselves are not mutated —
+# construction time, so they need a forced rebuild to pick up the relaxation. The wrappers themselves are not mutated -
 # their own field annotations stay as-is.
 _REBUILT_RESPONSE_WRAPPERS = (
     'ListOfActorsInStoreResponse',
@@ -138,9 +145,9 @@ def _relax_item_validation() -> Any:
     """Relax only the element type of `items` on paginated list models for the test run.
 
     Pagination tests feed synthetic `{'id': N}` items that don't satisfy the real API schemas (`ActorShort`,
-    `BuildShort`, `Request`, `EnvVar`, …). Instead of bypassing validation wholesale, each inner `ListOf*` model has its
-    `items` field swapped to `list[dict]` and rebuilt. Outer `.data` wrapping and every pagination-metadata field remain
-    validated.
+    `BuildShort`, `Request`, `EnvVar`, ...). Instead of bypassing validation wholesale, each inner `ListOf*` model
+    has its `items` field swapped to `list[dict]` and rebuilt. Outer `.data` wrapping and every pagination-metadata
+    field remain validated.
     """
     relaxed_field = FieldInfo.from_annotation(list[dict])
     originals: dict[type[BaseModel], FieldInfo] = {}
@@ -171,7 +178,7 @@ def create_items(start: int, end: int, step: int | None = None) -> list[dict[str
 
 
 def _is_true(value: str | None) -> bool:
-    """Match the `'true'` wire form produced by the client's bool→string serialization."""
+    """Match the `'true'` wire form produced by the client's bool->string serialization."""
     return value == 'true'
 
 
@@ -235,7 +242,7 @@ def _handle_cursor_pagination(request: Request) -> Response:
     """Serve a cursor-paginated Apify API response for KVS keys and RQ requests.
 
     Holds 2500 synthetic items whose integer `id` equals their position. Each page is capped at 1000 items. KVS uses
-    `exclusiveStartKey`; RQ uses the opaque `cursor`. Both values encode the last-seen item id as a string — the
+    `exclusiveStartKey`; RQ uses the opaque `cursor`. Both values encode the last-seen item id as a string - the
     next page starts at id + 1.
     """
     params = request.args
@@ -617,3 +624,64 @@ 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
+
+
+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(*, offset: int | None = None, **_kwargs: object) -> FakeOffsetPage:
+        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(*, offset: int | None = None, **_kwargs: object) -> FakeOffsetPage:
+        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=[]`) with a live cursor must not stop the cursor iterator."""
+    pages = {
+        None: ListOfRequests(items=[], limit=1000, next_cursor='c1'),
+        'c1': ListOfRequests(items=[{'id': 1}, {'id': 2}], limit=1000, next_cursor=None),
+    }
+
+    def callback(*, cursor: str | None = None, **_kwargs: object) -> ListOfRequests:
+        return pages[cursor]
+
+    assert list(get_cursor_iterator(callback, chunk_size=1000)) == [{'id': 1}, {'id': 2}]
+
+
+async def test_cursor_iterator_async_continues_past_fully_filtered_page() -> None:
+    """A fully-filtered page (`items=[]`) with a live cursor must not stop the async cursor iterator."""
+    pages = {
+        None: ListOfRequests(items=[], limit=1000, next_cursor='c1'),
+        'c1': ListOfRequests(items=[{'id': 1}, {'id': 2}], limit=1000, next_cursor=None),
+    }
+
+    async def callback(*, cursor: str | None = None, **_kwargs: object) -> ListOfRequests:
+        return pages[cursor]
+
+    collected = [item async for item in get_cursor_iterator_async(callback, chunk_size=1000)]
+    assert collected == [{'id': 1}, {'id': 2}]