Skip to content

Commit d382995

Browse files
authored
feat: add Actor.child_runs with clients for named child runs (#1151)
`Actor.child_runs` is a sync property that maps each run name to a `RunClientAsync` for the current run under that name. It has the same shape as `Actor.childRuns` in the JS SDK (apify/apify-sdk-js#759). The client covers the run and its storages: ```python run = await Actor.child_runs['scrape-eu'].wait_for_finish() ``` The SDK reads the registry from the default key-value store when the Actor initializes, so runs recorded before a migration or resurrection are included and reading the property makes no API calls. That adds one record read to every init. A malformed record fails the init and leaves the Actor uninitialized. Runs started without a `run_name` aren't tracked. Runs that were replaced under a name stay in the `history` of the `__ACTOR_CHILD_RUNS` record and aren't exposed. A run started or reattached with a custom `token` gets a client with that token. A run recorded before a migration and not reattached since gets the default client, because the token isn't stored. It also fixes a duplicate `max_items` deprecation warning from #1149: a named start that resurrects its recorded run no longer warns a second time from inside the SDK. Docs are in #1160. Closes: #1129 *✍️ Drafted by Claude Code*
1 parent a036c2d commit d382995

5 files changed

Lines changed: 207 additions & 40 deletions

File tree

‎src/apify/_actor.py‎

Lines changed: 36 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,7 @@ async def __aenter__(self) -> Self:
182182
- Sets up local or cloud storage clients depending on whether the Actor runs locally or on the Apify platform.
183183
- Configures the event manager and starts periodic state persistence.
184184
- Initializes the charging manager for handling charging events.
185+
- Loads the named child runs recorded by an earlier attempt of this Actor run.
185186
- Configures logging after all core services are registered.
186187
187188
This method must be called exactly once per Actor instance. Re-initializing an Actor or having multiple
@@ -230,9 +231,20 @@ async def __aenter__(self) -> Self:
230231
# Mark initialization as complete and update global state.
231232
self._active = True
232233

233-
if not Actor.is_at_home():
234-
# Make sure that the input related KVS is initialized to ensure that the input aware client is used
235-
await self.open_key_value_store()
234+
try:
235+
if not Actor.is_at_home():
236+
# Make sure that the input related KVS is initialized to ensure that the input aware client is used
237+
await self.open_key_value_store()
238+
239+
await self._child_run_registry.load()
240+
except BaseException:
241+
# Undo the initialization, since a failed `__aenter__` gets no `__aexit__`.
242+
self._active = False
243+
try:
244+
await self._charging_manager_implementation.__aexit__(None, None, None)
245+
finally:
246+
await self.event_manager.__aexit__(None, None, None)
247+
raise
236248
return self
237249

238250
async def __aexit__(
@@ -384,6 +396,24 @@ def apify_client(self) -> ApifyClientAsync:
384396
self._apify_client = self.new_client()
385397
return self._apify_client
386398

399+
@property
400+
@_ensure_context
401+
def child_runs(self) -> dict[str, RunClientAsync]:
402+
"""Clients for the named child runs of this Actor run, keyed by the run name.
403+
404+
Every run started by `Actor.start`, `Actor.call`, `Actor.start_task` or `Actor.call_task` with a `run_name` is
405+
included, even one started before a migration or resurrection of this Actor run. Runs started without a
406+
`run_name` are not tracked. Each client points to the current run under its name:
407+
408+
```python
409+
run = await Actor.child_runs['my-child'].wait_for_finish()
410+
```
411+
412+
A run started or reattached with a custom `token` since the last migration or resurrection of this Actor run
413+
uses that token. Any other run uses the default client.
414+
"""
415+
return self._child_run_registry.run_clients(self.apify_client)
416+
387417
@cached_property
388418
def configuration(self) -> Configuration:
389419
"""Actor configuration, uses the default instance if not explicitly set."""
@@ -1026,7 +1056,6 @@ async def start(
10261056
run_input=run_input,
10271057
client=client,
10281058
start_run=start_run,
1029-
token=token,
10301059
build=build,
10311060
max_items=max_items,
10321061
max_total_charge_usd=max_total_charge_usd,
@@ -1230,7 +1259,6 @@ async def call(
12301259
force_permission_level=force_permission_level,
12311260
webhooks=to_client_representations(webhooks),
12321261
),
1233-
token=token,
12341262
build=build,
12351263
max_items=max_items,
12361264
max_total_charge_usd=max_total_charge_usd,
@@ -1257,7 +1285,6 @@ async def _find_or_start_child_run(
12571285
run_input: Any,
12581286
client: ApifyClientAsync,
12591287
start_run: Callable[[], Awaitable[Run]],
1260-
token: str | None,
12611288
build: str | None,
12621289
max_items: int | None,
12631290
max_total_charge_usd: Decimal | None,
@@ -1272,15 +1299,13 @@ async def _find_or_start_child_run(
12721299
run_input=run_input,
12731300
client=client,
12741301
start_run=start_run,
1275-
resurrect_run=partial(
1276-
self.resurrect,
1277-
token=token,
1302+
resurrect_run=lambda run_id: client.run(run_id).resurrect(
12781303
build=build,
1304+
memory_mbytes=memory_mbytes,
1305+
run_timeout=self._resolve_run_timeout(timeout),
12791306
max_items=max_items,
12801307
max_total_charge_usd=max_total_charge_usd,
12811308
restart_on_error=restart_on_error,
1282-
memory_mbytes=memory_mbytes,
1283-
timeout=timeout,
12841309
),
12851310
)
12861311

@@ -1391,7 +1416,6 @@ async def start_task(
13911416
run_input=task_input,
13921417
client=client,
13931418
start_run=start_run,
1394-
token=token,
13951419
build=build,
13961420
max_items=max_items,
13971421
max_total_charge_usd=max_total_charge_usd,
@@ -1492,7 +1516,6 @@ async def call_task(
14921516
run_timeout=self._resolve_run_timeout(timeout),
14931517
webhooks=to_client_representations(webhooks),
14941518
),
1495-
token=token,
14961519
build=build,
14971520
max_items=max_items,
14981521
max_total_charge_usd=max_total_charge_usd,

‎src/apify/_child_runs.py‎

Lines changed: 38 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,8 @@ def __init__(self, open_key_value_store: Callable[[], Awaitable[KeyValueStore]])
104104
"""Guards loading the records and writing them back to the key-value store."""
105105
self._name_locks: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary()
106106
"""Serializes `find_or_start` per name. A lock is dropped once no call under its name holds it."""
107+
self._clients: dict[str, ApifyClientAsync] = {}
108+
"""Client the run under each name was last started or looked up with. Lost on a migration."""
107109

108110
async def find_or_start(
109111
self,
@@ -140,18 +142,20 @@ async def find_or_start(
140142
records = await self._load()
141143
record = records.get(name)
142144

143-
if record is None:
144-
run = await self._start(name, checksum=checksum, start_run=start_run, history=[])
145-
return run, True
146-
147-
if record.checksum != checksum:
145+
if record is not None and record.checksum != checksum:
148146
raise ValueError(
149147
f'The run name "{name}" was already used for a different Actor, task or input. '
150148
'Use a unique `run_name` for each child run.'
151149
)
152150

151+
if record is None:
152+
run = await self._start(name, checksum=checksum, start_run=start_run, history=[])
153+
self._clients[name] = client
154+
return run, True
155+
153156
run_client = client.run(record.run_id)
154157
run = await _get_recorded_run(run_client)
158+
self._clients[name] = client
155159

156160
if run is not None and run.status in _SETTLING_STATUSES:
157161
run = await run_client.wait_for_finish()
@@ -176,6 +180,21 @@ async def find_or_start(
176180
await self.update(name, run)
177181
return run, False
178182

183+
def run_clients(self, default_client: ApifyClientAsync) -> dict[str, RunClientAsync]:
184+
"""Return a client for the current run under each recorded name.
185+
186+
Each client comes from the client its name was last started or looked up with in this process, so a run started
187+
with a custom token uses that token.
188+
189+
Args:
190+
default_client: Client used for a name not started or looked up in this process, e.g. one recorded before a
191+
migration.
192+
"""
193+
return {
194+
name: self._clients.get(name, default_client).run(record.run_id)
195+
for name, record in (self._records or {}).items()
196+
}
197+
179198
async def update(self, name: str, run: Run) -> None:
180199
"""Record the latest observed status of the run recorded under `name`.
181200
@@ -203,20 +222,23 @@ async def _start(
203222
await self._save(name, record)
204223
return run
205224

206-
async def _load(self) -> dict[str, ChildRunRecord]:
225+
async def load(self) -> dict[str, ChildRunRecord]:
226+
"""Read the records from the default key-value store, replacing any read before."""
207227
async with self._lock:
208-
if self._records is None:
209-
key_value_store = await self._open_key_value_store()
210-
stored = await key_value_store.get_value(CHILD_RUNS_KEY)
211-
try:
212-
self._records = _records_adapter.validate_python(stored or {})
213-
except ValidationError as exc:
214-
raise ValueError(
215-
f'The child run registry under the "{CHILD_RUNS_KEY}" key in the default key-value store '
216-
'is malformed.'
217-
) from exc
228+
key_value_store = await self._open_key_value_store()
229+
stored = await key_value_store.get_value(CHILD_RUNS_KEY)
230+
try:
231+
self._records = _records_adapter.validate_python(stored or {})
232+
except ValidationError as exc:
233+
raise ValueError(
234+
f'The child run registry under the "{CHILD_RUNS_KEY}" key in the default key-value store '
235+
'is malformed.'
236+
) from exc
218237
return self._records
219238

239+
async def _load(self) -> dict[str, ChildRunRecord]:
240+
return self._records if self._records is not None else await self.load()
241+
220242
async def _save(self, name: str, record: ChildRunRecord) -> None:
221243
records = await self._load()
222244
key_value_store = await self._open_key_value_store()

‎tests/e2e/test_actor_child_runs.py‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ async def test_named_child_run_is_reattached_after_reboot(
1313
make_actor: MakeActorFunction,
1414
run_actor: RunActorFunction,
1515
) -> None:
16-
"""A named child run started before a reboot is reattached and awaited by a named call after it."""
16+
"""A named child run started before a reboot is listed after it and reattached by a named call."""
1717

1818
async def main() -> None:
1919
async with Actor:
@@ -31,6 +31,10 @@ async def main() -> None:
3131
await Actor.reboot()
3232
return
3333

34+
child_runs = Actor.child_runs
35+
assert child_runs.keys() == {'child'}, f'child_runs={child_runs}'
36+
assert child_runs['child'].resource_id == child_run_id, f'child_runs={child_runs}'
37+
3438
run = await Actor.call(actor_id=actor_id, run_input={'is_child': True}, run_name='child')
3539
assert run is not None, 'run is None'
3640
assert run.id == child_run_id, f'run.id={run.id}, child_run_id={child_run_id}'

0 commit comments

Comments
 (0)