diff --git a/src/apify_client/_pagination.py b/src/apify_client/_pagination.py index 3433c242..4facbe58 100644 --- a/src/apify_client/_pagination.py +++ b/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 diff --git a/tests/unit/test_client_pagination.py b/tests/unit/test_client_pagination.py index 5bf4e4a2..ac9d92b5 100644 --- a/tests/unit/test_client_pagination.py +++ b/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}]