Skip to content

Commit 2e7f75c

Browse files
authored
fix: Keep pagination iterators advancing past fully-filtered pages (#964)
The offset- and cursor-based pagination iterators stopped as soon as a page's `items` list came back empty. But under dataset filters (`clean`, `skip_empty`, `skip_hidden`), the API can *scan* a full chunk (up to 1000 items) while *returning* none — `items=[]` with `x-apify-pagination-count: 1000`. The iterators broke on that page and silently never yielded any item behind it, even though the offset bookkeeping (`max(count, len(items))`) had already advanced past the scanned rows. Now all four iterators (offset + cursor, sync + async) terminate on the number of items *scanned* rather than *returned*: they keep advancing the offset/cursor across fully-filtered pages and stop only when a page scans nothing. The `limit` and `cursor is None` checks are unchanged. Behavior is identical for every response without a divergent `count` (collection endpoints, `ListOfRequests`). Added a regression test for each of the four iterators. Related to apify/apify-core#27324. *✍️ Drafted by Claude Code*
1 parent 9bdca6a commit 2e7f75c

2 files changed

Lines changed: 95 additions & 19 deletions

File tree

src/apify_client/_pagination.py

Lines changed: 21 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919
class HasItems(Protocol[T]):
2020
"""Structural contract for a single page of results from a paginated API endpoint.
2121
22-
Implementations must expose `items`. They may optionally expose `count` the number of items scanned by the API for
22+
Implementations must expose `items`. They may optionally expose `count` - the number of items scanned by the API for
2323
this page, which can exceed `len(items)` when filters drop items from the response. The iterator helpers consult
2424
`count` opportunistically via `getattr` for offset bookkeeping and fall back to `len(items)` when it is absent.
2525
"""
@@ -41,8 +41,11 @@ def get_items_iterator(
4141
is used for offset bookkeeping (the Apify API's `count` reflects items scanned, which can exceed items returned when
4242
filters are applied).
4343
44-
Iteration stops when a page returns no items or when the user-requested `limit` is reached. The `total` field is
45-
intentionally not consulted, because it can change between calls.
44+
Iteration stops when a page scans no items (`count` is `0`, or `items` is empty when `count` is absent) or when the
45+
user-requested `limit` is reached. A page can scan items while returning none - filters like `clean` drop items from
46+
`items` but still count toward `count` - so terminating on scanned rather than returned items keeps the iterator
47+
advancing across fully-filtered pages. The `total` field is intentionally not consulted, because it can change
48+
between calls.
4649
4750
Args:
4851
callback: Function returning a single page of items.
@@ -62,9 +65,10 @@ def get_items_iterator(
6265
)
6366
yield from current_page.items
6467

65-
fetched_items += max(getattr(current_page, 'count', 0), len(current_page.items))
68+
page_scanned = max(getattr(current_page, 'count', 0), len(current_page.items))
69+
fetched_items += page_scanned
6670

67-
if not current_page.items or (initial_limit and fetched_items >= initial_limit):
71+
if not page_scanned or (initial_limit and fetched_items >= initial_limit):
6872
break
6973

7074

@@ -92,9 +96,10 @@ async def get_items_iterator_async(
9296
for item in current_page.items:
9397
yield item
9498

95-
fetched_items += max(getattr(current_page, 'count', 0), len(current_page.items))
99+
page_scanned = max(getattr(current_page, 'count', 0), len(current_page.items))
100+
fetched_items += page_scanned
96101

97-
if not current_page.items or (initial_limit and fetched_items >= initial_limit):
102+
if not page_scanned or (initial_limit and fetched_items >= initial_limit):
98103
break
99104

100105

@@ -124,8 +129,11 @@ def get_cursor_iterator(
124129
"""Yield individual items from cursor-paginated API responses.
125130
126131
Cursor pagination is restricted to the two API responses that expose it: `ListOfKeys` (for key-value store keys) and
127-
`ListOfRequests` (for request queue requests). Iteration ends when a page returns no items, the next cursor is
128-
`None`, or the user-requested `limit` is reached.
132+
`ListOfRequests` (for request queue requests). Iteration ends when the next cursor is `None` or the user-requested
133+
`limit` is reached. Emptiness alone does not stop iteration: server-side filters (such as the request-queue state
134+
`filter`) can drop every item on a page while a live cursor still points at more data, so termination relies on the
135+
cursor, not on whether a page returned items. Unlike offset responses, cursor responses expose no scanned-item
136+
`count`, so `count` cannot be used to detect a fully-filtered page here.
129137
130138
Args:
131139
callback: Function returning a single page of items. Receives `cursor` and `limit` kwargs.
@@ -144,12 +152,12 @@ def get_cursor_iterator(
144152
)
145153
yield from current_page.items
146154

147-
fetched_items += max(getattr(current_page, 'count', 0), len(current_page.items))
155+
fetched_items += len(current_page.items)
148156
cursor = (
149157
current_page.next_exclusive_start_key if isinstance(current_page, ListOfKeys) else current_page.next_cursor
150158
)
151159

152-
if not current_page.items or cursor is None or (initial_limit and fetched_items >= initial_limit):
160+
if cursor is None or (initial_limit and fetched_items >= initial_limit):
153161
break
154162

155163

@@ -189,12 +197,12 @@ async def get_cursor_iterator_async(
189197
for item in current_page.items:
190198
yield item
191199

192-
fetched_items += max(getattr(current_page, 'count', 0), len(current_page.items))
200+
fetched_items += len(current_page.items)
193201
cursor = (
194202
current_page.next_exclusive_start_key if isinstance(current_page, ListOfKeys) else current_page.next_cursor
195203
)
196204

197-
if not current_page.items or cursor is None or (initial_limit and fetched_items >= initial_limit):
205+
if cursor is None or (initial_limit and fetched_items >= initial_limit):
198206
break
199207

200208

tests/unit/test_client_pagination.py

Lines changed: 74 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,13 @@
1111

1212
from apify_client import ApifyClient, ApifyClientAsync
1313
from apify_client import _models as _models_module
14+
from apify_client._models import ListOfRequests
15+
from apify_client._pagination import (
16+
get_cursor_iterator,
17+
get_cursor_iterator_async,
18+
get_items_iterator,
19+
get_items_iterator_async,
20+
)
1421
from apify_client._resource_clients import (
1522
ActorCollectionClient,
1623
ActorCollectionClientAsync,
@@ -113,7 +120,7 @@
113120
)
114121

115122
# Outer wrappers that embed a relaxed list model via `.data`. Their compiled schema pins the inner's schema at
116-
# construction time, so they need a forced rebuild to pick up the relaxation. The wrappers themselves are not mutated
123+
# construction time, so they need a forced rebuild to pick up the relaxation. The wrappers themselves are not mutated -
117124
# their own field annotations stay as-is.
118125
_REBUILT_RESPONSE_WRAPPERS = (
119126
'ListOfActorsInStoreResponse',
@@ -138,9 +145,9 @@ def _relax_item_validation() -> Any:
138145
"""Relax only the element type of `items` on paginated list models for the test run.
139146
140147
Pagination tests feed synthetic `{'id': N}` items that don't satisfy the real API schemas (`ActorShort`,
141-
`BuildShort`, `Request`, `EnvVar`, ). Instead of bypassing validation wholesale, each inner `ListOf*` model has its
142-
`items` field swapped to `list[dict]` and rebuilt. Outer `.data` wrapping and every pagination-metadata field remain
143-
validated.
148+
`BuildShort`, `Request`, `EnvVar`, ...). Instead of bypassing validation wholesale, each inner `ListOf*` model
149+
has its `items` field swapped to `list[dict]` and rebuilt. Outer `.data` wrapping and every pagination-metadata
150+
field remain validated.
144151
"""
145152
relaxed_field = FieldInfo.from_annotation(list[dict])
146153
originals: dict[type[BaseModel], FieldInfo] = {}
@@ -171,7 +178,7 @@ def create_items(start: int, end: int, step: int | None = None) -> list[dict[str
171178

172179

173180
def _is_true(value: str | None) -> bool:
174-
"""Match the `'true'` wire form produced by the client's boolstring serialization."""
181+
"""Match the `'true'` wire form produced by the client's bool->string serialization."""
175182
return value == 'true'
176183

177184

@@ -235,7 +242,7 @@ def _handle_cursor_pagination(request: Request) -> Response:
235242
"""Serve a cursor-paginated Apify API response for KVS keys and RQ requests.
236243
237244
Holds 2500 synthetic items whose integer `id` equals their position. Each page is capped at 1000 items. KVS uses
238-
`exclusiveStartKey`; RQ uses the opaque `cursor`. Both values encode the last-seen item id as a string the
245+
`exclusiveStartKey`; RQ uses the opaque `cursor`. Both values encode the last-seen item id as a string - the
239246
next page starts at id + 1.
240247
"""
241248
params = request.args
@@ -617,3 +624,64 @@ async def test_rq_list_requests_iterable_async(
617624
client: RequestQueueClientAsync = _CLIENT_FACTORIES[client_name](_make_async_client(pagination_server))
618625
returned_items = [dict(item) async for item in client.iterate_requests(**inputs)]
619626
assert returned_items == expected_items
627+
628+
629+
class FakeOffsetPage:
630+
"""Offset-paginated page whose `count` (items scanned) may exceed `len(items)` when filters drop items."""
631+
632+
def __init__(self, items: list[dict[str, int]], count: int) -> None:
633+
self.items = items
634+
self.count = count
635+
636+
637+
def test_items_iterator_continues_past_fully_filtered_page() -> None:
638+
"""A fully-filtered page (`items=[]`, `count>0`) must not stop the offset iterator while more data was scanned."""
639+
pages = {
640+
0: FakeOffsetPage(items=[], count=1000),
641+
1000: FakeOffsetPage(items=[{'id': 1}, {'id': 2}], count=2),
642+
}
643+
644+
def callback(*, offset: int | None = None, **_kwargs: object) -> FakeOffsetPage:
645+
return pages.get(offset or 0, FakeOffsetPage(items=[], count=0))
646+
647+
assert list(get_items_iterator(callback, chunk_size=1000)) == [{'id': 1}, {'id': 2}]
648+
649+
650+
async def test_items_iterator_async_continues_past_fully_filtered_page() -> None:
651+
"""A fully-filtered page (`items=[]`, `count>0`) must not stop the async offset iterator while more was scanned."""
652+
pages = {
653+
0: FakeOffsetPage(items=[], count=1000),
654+
1000: FakeOffsetPage(items=[{'id': 1}, {'id': 2}], count=2),
655+
}
656+
657+
async def callback(*, offset: int | None = None, **_kwargs: object) -> FakeOffsetPage:
658+
return pages.get(offset or 0, FakeOffsetPage(items=[], count=0))
659+
660+
assert [item async for item in get_items_iterator_async(callback, chunk_size=1000)] == [{'id': 1}, {'id': 2}]
661+
662+
663+
def test_cursor_iterator_continues_past_fully_filtered_page() -> None:
664+
"""A fully-filtered page (`items=[]`) with a live cursor must not stop the cursor iterator."""
665+
pages = {
666+
None: ListOfRequests(items=[], limit=1000, next_cursor='c1'),
667+
'c1': ListOfRequests(items=[{'id': 1}, {'id': 2}], limit=1000, next_cursor=None),
668+
}
669+
670+
def callback(*, cursor: str | None = None, **_kwargs: object) -> ListOfRequests:
671+
return pages[cursor]
672+
673+
assert list(get_cursor_iterator(callback, chunk_size=1000)) == [{'id': 1}, {'id': 2}]
674+
675+
676+
async def test_cursor_iterator_async_continues_past_fully_filtered_page() -> None:
677+
"""A fully-filtered page (`items=[]`) with a live cursor must not stop the async cursor iterator."""
678+
pages = {
679+
None: ListOfRequests(items=[], limit=1000, next_cursor='c1'),
680+
'c1': ListOfRequests(items=[{'id': 1}, {'id': 2}], limit=1000, next_cursor=None),
681+
}
682+
683+
async def callback(*, cursor: str | None = None, **_kwargs: object) -> ListOfRequests:
684+
return pages[cursor]
685+
686+
collected = [item async for item in get_cursor_iterator_async(callback, chunk_size=1000)]
687+
assert collected == [{'id': 1}, {'id': 2}]

0 commit comments

Comments
 (0)