Skip to content

Commit df73e6e

Browse files
vdusekvdusek
committed
Add async iterators for paginated resources
1 parent 9c456f7 commit df73e6e

13 files changed

Lines changed: 648 additions & 61 deletions

‎src/apify_client/_resource_clients/actor_collection.py‎

Lines changed: 98 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,11 @@
11
from __future__ import annotations
22

3-
from typing import Any, Literal
3+
from typing import TYPE_CHECKING, Any, Literal
44

5-
from apify_client._models import Actor, CreateActorResponse, GetListOfActorsResponse, ListOfActors
5+
from apify_client._models import Actor, ActorShort, CreateActorResponse, GetListOfActorsResponse, ListOfActors
6+
7+
if TYPE_CHECKING:
8+
from collections.abc import AsyncIterator, Iterator
69
from apify_client._resource_clients._resource_client import ResourceClient, ResourceClientAsync
710
from apify_client._resource_clients.actor import get_actor_representation
811
from apify_client._utils import filter_none_values, response_to_dict
@@ -46,6 +49,52 @@ def list(
4649
response_as_dict = response_to_dict(response)
4750
return GetListOfActorsResponse.model_validate(response_as_dict).data
4851

52+
def iterate(
53+
self,
54+
*,
55+
my: bool | None = None,
56+
limit: int | None = None,
57+
desc: bool | None = None,
58+
) -> Iterator[ActorShort]:
59+
"""Iterate over the Actors the user has created or used.
60+
61+
https://docs.apify.com/api/v2#/reference/actors/actor-collection/get-list-of-actors
62+
63+
Args:
64+
my: If True, will return only Actors which the user has created themselves.
65+
limit: Maximum number of Actors to return. By default there is no limit.
66+
desc: Whether to sort the Actors in descending order based on their creation date.
67+
68+
Yields:
69+
An Actor from the collection.
70+
"""
71+
cache_size = 1000
72+
read_items = 0
73+
offset = 0
74+
75+
while True:
76+
effective_limit = cache_size
77+
if limit is not None:
78+
if read_items == limit:
79+
break
80+
effective_limit = min(cache_size, limit - read_items)
81+
82+
current_page = self.list(
83+
my=my,
84+
limit=effective_limit,
85+
offset=offset,
86+
desc=desc,
87+
)
88+
89+
yield from current_page.items
90+
91+
current_page_item_count = len(current_page.items)
92+
read_items += current_page_item_count
93+
offset += current_page_item_count
94+
95+
if current_page_item_count < cache_size:
96+
break
97+
4998
def create(
5099
self,
51100
*,
@@ -185,6 +234,53 @@ async def list(
185234
response_as_dict = response_to_dict(response)
186235
return GetListOfActorsResponse.model_validate(response_as_dict).data
187236

237+
async def iterate(
238+
self,
239+
*,
240+
my: bool | None = None,
241+
limit: int | None = None,
242+
desc: bool | None = None,
243+
) -> AsyncIterator[ActorShort]:
244+
"""Iterate over the Actors the user has created or used.
245+
246+
https://docs.apify.com/api/v2#/reference/actors/actor-collection/get-list-of-actors
247+
248+
Args:
249+
my: If True, will return only Actors which the user has created themselves.
250+
limit: Maximum number of Actors to return. By default there is no limit.
251+
desc: Whether to sort the Actors in descending order based on their creation date.
252+
253+
Yields:
254+
An Actor from the collection.
255+
"""
256+
cache_size = 1000
257+
read_items = 0
258+
offset = 0
259+
260+
while True:
261+
effective_limit = cache_size
262+
if limit is not None:
263+
if read_items == limit:
264+
break
265+
effective_limit = min(cache_size, limit - read_items)
266+
267+
current_page = await self.list(
268+
my=my,
269+
limit=effective_limit,
270+
offset=offset,
271+
desc=desc,
272+
)
273+
274+
for item in current_page.items:
275+
yield item
276+
277+
current_page_item_count = len(current_page.items)
278+
read_items += current_page_item_count
279+
offset += current_page_item_count
280+
281+
if current_page_item_count < cache_size:
282+
break
283+
188284
async def create(
189285
self,
190286
*,

‎src/apify_client/_resource_clients/key_value_store.py‎

Lines changed: 106 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,13 @@
77
from urllib.parse import urlencode, urlparse, urlunparse
88

99
from apify_client._consts import FAST_OPERATION_TIMEOUT_SECS, STANDARD_OPERATION_TIMEOUT_SECS
10-
from apify_client._models import GetKeyValueStoreResponse, GetListOfKeysResponse, KeyValueStore, ListOfKeys
10+
from apify_client._models import (
11+
GetKeyValueStoreResponse,
12+
GetListOfKeysResponse,
13+
KeyValueStore,
14+
KeyValueStoreKey,
15+
ListOfKeys,
16+
)
1117
from apify_client._resource_clients._resource_client import ResourceClient, ResourceClientAsync
1218
from apify_client._utils import (
1319
catch_not_found_or_throw,
@@ -172,6 +178,55 @@ def list_keys(
172178
result = response.json()
173179
return GetListOfKeysResponse.model_validate(result).data
174180

181+
def iterate_keys(
182+
self,
183+
*,
184+
limit: int | None = None,
185+
collection: str | None = None,
186+
prefix: str | None = None,
187+
signature: str | None = None,
188+
) -> Iterator[KeyValueStoreKey]:
189+
"""Iterate over the keys in the key-value store.
190+
191+
https://docs.apify.com/api/v2#/reference/key-value-stores/key-collection/get-list-of-keys
192+
193+
Args:
194+
limit: Maximum number of keys to return. By default there is no limit.
195+
collection: The name of the collection in store schema to list keys from.
196+
prefix: The prefix of the keys to be listed.
197+
signature: Signature used to access the items.
198+
199+
Yields:
200+
A key from the key-value store.
201+
"""
202+
cache_size = 1000
203+
read_keys = 0
204+
exclusive_start_key: str | None = None
205+
206+
while True:
207+
effective_limit = cache_size
208+
if limit is not None:
209+
if read_keys == limit:
210+
break
211+
effective_limit = min(cache_size, limit - read_keys)
212+
213+
current_keys_page = self.list_keys(
214+
limit=effective_limit,
215+
exclusive_start_key=exclusive_start_key,
216+
collection=collection,
217+
prefix=prefix,
218+
signature=signature,
219+
)
220+
221+
yield from current_keys_page.items
222+
223+
read_keys += len(current_keys_page.items)
224+
225+
if not current_keys_page.is_truncated:
226+
break
227+
228+
exclusive_start_key = current_keys_page.next_exclusive_start_key
229+
175230
def get_record(self, key: str, signature: str | None = None) -> dict | None:
176231
"""Retrieve the given record from the key-value store.
177232
@@ -529,6 +584,56 @@ async def list_keys(
529584
result = response.json()
530585
return GetListOfKeysResponse.model_validate(result).data
531586

587+
async def iterate_keys(
588+
self,
589+
*,
590+
limit: int | None = None,
591+
collection: str | None = None,
592+
prefix: str | None = None,
593+
signature: str | None = None,
594+
) -> AsyncIterator[KeyValueStoreKey]:
595+
"""Iterate over the keys in the key-value store.
596+
597+
https://docs.apify.com/api/v2#/reference/key-value-stores/key-collection/get-list-of-keys
598+
599+
Args:
600+
limit: Maximum number of keys to return. By default there is no limit.
601+
collection: The name of the collection in store schema to list keys from.
602+
prefix: The prefix of the keys to be listed.
603+
signature: Signature used to access the items.
604+
605+
Yields:
606+
A key from the key-value store.
607+
"""
608+
cache_size = 1000
609+
read_keys = 0
610+
exclusive_start_key: str | None = None
611+
612+
while True:
613+
effective_limit = cache_size
614+
if limit is not None:
615+
if read_keys == limit:
616+
break
617+
effective_limit = min(cache_size, limit - read_keys)
618+
619+
current_keys_page = await self.list_keys(
620+
limit=effective_limit,
621+
exclusive_start_key=exclusive_start_key,
622+
collection=collection,
623+
prefix=prefix,
624+
signature=signature,
625+
)
626+
627+
for key in current_keys_page.items:
628+
yield key
629+
630+
read_keys += len(current_keys_page.items)
631+
632+
if not current_keys_page.is_truncated:
633+
break
634+
635+
exclusive_start_key = current_keys_page.next_exclusive_start_key
636+
532637
async def get_record(self, key: str, signature: str | None = None) -> dict | None:
533638
"""Retrieve the given record from the key-value store.
534639

‎src/apify_client/_resource_clients/run_collection.py‎

Lines changed: 115 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,12 @@
22

33
from typing import TYPE_CHECKING, Any
44

5-
from apify_client._models import GetListOfRunsResponse, ListOfRuns
5+
from apify_client._models import GetListOfRunsResponse, ListOfRuns, RunShort
66
from apify_client._resource_clients._resource_client import ResourceClient, ResourceClientAsync
77
from apify_client._utils import enum_to_value, response_to_dict
88

99
if TYPE_CHECKING:
10+
from collections.abc import AsyncIterator, Iterator
1011
from datetime import datetime
1112

1213
from apify_client._consts import ActorJobStatus
@@ -65,6 +66,62 @@ def list(
6566
response_as_dict = response_to_dict(response)
6667
return GetListOfRunsResponse.model_validate(response_as_dict).data
6768

69+
def iterate(
70+
self,
71+
*,
72+
limit: int | None = None,
73+
desc: bool | None = None,
74+
status: ActorJobStatus | list[ActorJobStatus] | None = None, # ty: ignore[invalid-type-form]
75+
started_before: str | datetime | None = None,
76+
started_after: str | datetime | None = None,
77+
) -> Iterator[RunShort]:
78+
"""Iterate over all Actor runs.
79+
80+
Iterate over all Actor runs, either of a single Actor, or all user's Actors, depending on where this client
81+
was initialized from.
82+
83+
https://docs.apify.com/api/v2#/reference/actors/run-collection/get-list-of-runs
84+
https://docs.apify.com/api/v2#/reference/actor-runs/run-collection/get-user-runs-list
85+
86+
Args:
87+
limit: Maximum number of runs to return. By default there is no limit.
88+
desc: Whether to sort the runs in descending order based on their start date.
89+
status: Retrieve only runs with the provided statuses.
90+
started_before: Only return runs started before this date (inclusive).
91+
started_after: Only return runs started after this date (inclusive).
92+
93+
Yields:
94+
A run from the collection.
95+
"""
96+
cache_size = 1000
97+
read_items = 0
98+
offset = 0
99+
100+
while True:
101+
effective_limit = cache_size
102+
if limit is not None:
103+
if read_items == limit:
104+
break
105+
effective_limit = min(cache_size, limit - read_items)
106+
107+
current_page = self.list(
108+
limit=effective_limit,
109+
offset=offset,
110+
desc=desc,
111+
status=status,
112+
started_before=started_before,
113+
started_after=started_after,
114+
)
115+
116+
yield from current_page.items
117+
118+
current_page_item_count = len(current_page.items)
119+
read_items += current_page_item_count
120+
offset += current_page_item_count
121+
122+
if current_page_item_count < cache_size:
123+
break
124+
68125

69126
class RunCollectionClientAsync(ResourceClientAsync):
70127
"""Async sub-client for listing Actor runs."""
@@ -118,3 +175,60 @@ async def list(
118175
)
119176
response_as_dict = response_to_dict(response)
120177
return GetListOfRunsResponse.model_validate(response_as_dict).data
178+
179+
async def iterate(
180+
self,
181+
*,
182+
limit: int | None = None,
183+
desc: bool | None = None,
184+
status: ActorJobStatus | list[ActorJobStatus] | None = None, # ty: ignore[invalid-type-form]
185+
started_before: str | datetime | None = None,
186+
started_after: str | datetime | None = None,
187+
) -> AsyncIterator[RunShort]:
188+
"""Iterate over all Actor runs.
189+
190+
Iterate over all Actor runs, either of a single Actor, or all user's Actors, depending on where this client
191+
was initialized from.
192+
193+
https://docs.apify.com/api/v2#/reference/actors/run-collection/get-list-of-runs
194+
https://docs.apify.com/api/v2#/reference/actor-runs/run-collection/get-user-runs-list
195+
196+
Args:
197+
limit: Maximum number of runs to return. By default there is no limit.
198+
desc: Whether to sort the runs in descending order based on their start date.
199+
status: Retrieve only runs with the provided statuses.
200+
started_before: Only return runs started before this date (inclusive).
201+
started_after: Only return runs started after this date (inclusive).
202+
203+
Yields:
204+
A run from the collection.
205+
"""
206+
cache_size = 1000
207+
read_items = 0
208+
offset = 0
209+
210+
while True:
211+
effective_limit = cache_size
212+
if limit is not None:
213+
if read_items == limit:
214+
break
215+
effective_limit = min(cache_size, limit - read_items)
216+
217+
current_page = await self.list(
218+
limit=effective_limit,
219+
offset=offset,
220+
desc=desc,
221+
status=status,
222+
started_before=started_before,
223+
started_after=started_after,
224+
)
225+
226+
for item in current_page.items:
227+
yield item
228+
229+
current_page_item_count = len(current_page.items)
230+
read_items += current_page_item_count
231+
offset += current_page_item_count
232+
233+
if current_page_item_count < cache_size:
234+
break

0 commit comments

Comments
 (0)