Merge branch 'feat/named-child-runs' into feat/child-runs-api

Author: vdusek

src/apify/_actor.py

@@ -946,7 +946,7 @@ async def start(
         timeout: timedelta | Literal['inherit'] | None = None,
         force_permission_level: ActorPermissionLevel | None = None,
         webhooks: list[Webhook] | None = None,
-        name: str | None = None,
+        run_name: str | None = None,
     ) -> Run:
         """Run an Actor on the Apify platform.
 
@@ -973,12 +973,12 @@ async def start(
             webhooks: Optional ad-hoc webhooks (https://docs.apify.com/webhooks/ad-hoc-webhooks) associated with
                 the Actor run which can be used to receive a notification, e.g. when the Actor finished or failed.
                 If you already have a webhook set up for the Actor or task, you do not have to add it again here.
-            name: Optional name of the child run, unique within this Actor run. A named run is recorded in the
+            run_name: Optional name of the child run, unique within this Actor run. A named run is recorded in the
                 default key-value store, so after a migration or resurrection of this Actor the same call reattaches
                 to the recorded run. A `SUCCEEDED` run is returned as is, an `ABORTED` or `TIMED-OUT` one is
                 resurrected, and a new run is started only when nothing is recorded under the name, or the recorded
                 run `FAILED` or no longer exists. The name is bound to `actor_id` exactly as passed, so reusing it with
-                any other value raises a `ValueError`.
+                any other value, or for a task, raises a `ValueError`.
 
         Returns:
             Info about the started Actor run
@@ -1008,11 +1008,11 @@ async def start(
             webhooks=to_client_representations(webhooks),
         )
 
-        if name is None:
+        if run_name is None:
             return await start_run()
 
         run, _ = await self._find_or_start_child_run(
-            name,
+            run_name,
             actor_id=actor_id,
             client=client,
             start_run=start_run,
@@ -1078,7 +1078,7 @@ async def call(
         webhooks: list[Webhook] | None = None,
         wait: timedelta | None = None,
         logger: logging.Logger | Literal['default'] | None = 'default',
-        name: str | None = None,
+        run_name: str | None = None,
     ) -> Run:
         """Start an Actor on the Apify Platform and wait for it to finish before returning.
 
@@ -1108,12 +1108,12 @@ async def call(
             logger: Logger used to redirect logs from the Actor run. Using "default" literal means that a predefined
                 default logger will be used. Setting `None` will disable any log propagation. Passing custom logger
                 will redirect logs to the provided logger.
-            name: Optional name of the child run, unique within this Actor run. A named run is recorded in the
+            run_name: Optional name of the child run, unique within this Actor run. A named run is recorded in the
                 default key-value store, so after a migration or resurrection of this Actor the same call reattaches
                 to the recorded run. A `SUCCEEDED` run is returned as is, an `ABORTED` or `TIMED-OUT` one is
                 resurrected, and a new run is started only when nothing is recorded under the name, or the recorded
                 run `FAILED` or no longer exists. The name is bound to `actor_id` exactly as passed, so reusing it with
-                any other value raises a `ValueError`.
+                any other value, or for a task, raises a `ValueError`.
 
         Returns:
             Info about the started Actor run.
@@ -1131,7 +1131,7 @@ async def call(
 
         actor_client = client.actor(actor_id)
 
-        if name is None:
+        if run_name is None:
             run = await actor_client.call(
                 run_input=run_input,
                 content_type=content_type,
@@ -1147,7 +1147,7 @@ async def call(
             )
         else:
             started_run, is_new = await self._find_or_start_child_run(
-                name,
+                run_name,
                 actor_id=actor_id,
                 client=client,
                 start_run=partial(
@@ -1182,7 +1182,8 @@ async def _find_or_start_child_run(
         self,
         name: str,
         *,
-        actor_id: str,
+        actor_id: str | None = None,
+        task_id: str | None = None,
         client: ApifyClientAsync,
         start_run: Callable[[], Awaitable[Run]],
         build: str | None,
@@ -1194,6 +1195,7 @@ async def _find_or_start_child_run(
         return await self._child_run_registry.find_or_start(
             name,
             actor_id=actor_id,
+            task_id=task_id,
             client=client,
             start_run=start_run,
             resurrect_run=lambda run_client: run_client.resurrect(
@@ -1253,6 +1255,7 @@ async def call_task(
         webhooks: list[Webhook] | None = None,
         wait: timedelta | None = None,
         token: str | None = None,
+        run_name: str | None = None,
     ) -> Run:
         """Start an Actor task on the Apify Platform and wait for it to finish before returning.
 
@@ -1278,6 +1281,12 @@ async def call_task(
                 be used to receive a notification, e.g. when the Actor finished or failed. If you already have
                 a webhook set up for the Actor, you do not have to add it again here.
             wait: The maximum time the server waits for the run to finish. If not provided, waits indefinitely.
+            run_name: Optional name of the child run, unique within this Actor run. A named run is recorded in the
+                default key-value store, so after a migration or resurrection of this Actor the same call reattaches
+                to the recorded run. A `SUCCEEDED` run is returned as is, an `ABORTED` or `TIMED-OUT` one is
+                resurrected, and a new run is started only when nothing is recorded under the name, or the recorded
+                run `FAILED` or no longer exists. The name is bound to `task_id` exactly as passed, so reusing it with
+                any other value, or for an Actor, raises a `ValueError`.
 
         Returns:
             Info about the started Actor run.
@@ -1294,15 +1303,40 @@ async def call_task(
             raise ValueError(f'Invalid timeout {timeout!r}: expected `None`, `"inherit"`, or a `timedelta`.')
 
         task_client = client.task(task_id)
-        run = await task_client.call(
-            task_input=task_input,
-            build=build,
-            restart_on_error=restart_on_error,
-            memory_mbytes=memory_mbytes,
-            run_timeout=task_call_timeout,
-            webhooks=to_client_representations(webhooks),
-            wait_duration=wait,
-        )
+
+        if run_name is None:
+            run = await task_client.call(
+                task_input=task_input,
+                build=build,
+                restart_on_error=restart_on_error,
+                memory_mbytes=memory_mbytes,
+                run_timeout=task_call_timeout,
+                webhooks=to_client_representations(webhooks),
+                wait_duration=wait,
+            )
+        else:
+            started_run, _ = await self._find_or_start_child_run(
+                run_name,
+                task_id=task_id,
+                client=client,
+                start_run=partial(
+                    task_client.start,
+                    task_input=task_input,
+                    build=build,
+                    restart_on_error=restart_on_error,
+                    memory_mbytes=memory_mbytes,
+                    run_timeout=task_call_timeout,
+                    webhooks=to_client_representations(webhooks),
+                ),
+                build=build,
+                max_total_charge_usd=None,
+                restart_on_error=restart_on_error,
+                memory_mbytes=memory_mbytes,
+                run_timeout=task_call_timeout,
+            )
+            run = await self._wait_for_child_run(
+                client.run(started_run.id), started_run, wait=wait, logger=None, from_start=False
+            )
 
         if run is None:
             raise RuntimeError(f'Failed to call Task with ID "{task_id}".')

src/apify/_child_runs.py

@@ -4,9 +4,9 @@
 from collections import defaultdict
 from dataclasses import dataclass
 from logging import getLogger
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Self
 
-from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError
+from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError, model_validator
 from pydantic.alias_generators import to_camel
 
 from apify._utils import docs_group
@@ -36,15 +36,28 @@ class ChildRunRecord(BaseModel):
 
     model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
 
-    actor_id: str
-    """The Actor ID or name the child was started with, as the caller passed it."""
+    actor_id: str | None = None
+    """The Actor ID or name the child was started with, as the caller passed it, or `None` for a task run."""
+
+    task_id: str | None = None
+    """The task ID or name the child was started with, as the caller passed it, or `None` for an Actor run."""
 
     run_id: str
     """ID of the current run under this name."""
 
     previous_run_ids: list[str] = Field(default_factory=list)
     """IDs of earlier runs under this name that failed or went missing and were replaced by a new run, oldest first."""
 
+    @model_validator(mode='after')
+    def _check_started_from(self) -> Self:
+        if (self.actor_id is None) == (self.task_id is None):
+            raise ValueError('Exactly one of `actor_id` and `task_id` must be set.')
+        return self
+
+
+def _describe_started_from(actor_id: str | None, task_id: str | None) -> str:
+    return f'Actor "{actor_id}"' if actor_id is not None else f'task "{task_id}"'
+
 
 @docs_group('Actor')
 @dataclass(frozen=True)
@@ -85,7 +98,8 @@ async def find_or_start(
         self,
         name: str,
         *,
-        actor_id: str,
+        actor_id: str | None = None,
+        task_id: str | None = None,
         client: ApifyClientAsync,
         start_run: Callable[[], Awaitable[Run]],
         resurrect_run: Callable[[RunClientAsync], Awaitable[Run]],
@@ -99,8 +113,9 @@ async def find_or_start(
         Args:
             name: Name of the child run, unique within the parent run.
             actor_id: The Actor to start. It must match the Actor already recorded under `name`.
+            task_id: The task to start, in place of `actor_id`. It must match the task already recorded under `name`.
             client: Client used to look up and resurrect the recorded run.
-            start_run: Starts a new run of the Actor.
+            start_run: Starts a new run of the Actor or task.
             resurrect_run: Resurrects the recorded run, given its run client.
 
         Returns:
@@ -111,12 +126,16 @@ async def find_or_start(
             record = records.get(name)
 
             if record is None:
-                return await self._start(name, actor_id=actor_id, start_run=start_run, previous_run_ids=[]), True
+                run = await self._start(
+                    name, actor_id=actor_id, task_id=task_id, start_run=start_run, previous_run_ids=[]
+                )
+                return run, True
 
-            if record.actor_id != actor_id:
+            if (record.actor_id, record.task_id) != (actor_id, task_id):
                 raise ValueError(
-                    f'Child run "{name}" is already recorded for Actor "{record.actor_id}", '
-                    f'it cannot be reused for Actor "{actor_id}".'
+                    f'Child run "{name}" is already recorded for '
+                    f'{_describe_started_from(record.actor_id, record.task_id)}, '
+                    f'it cannot be reused for {_describe_started_from(actor_id, task_id)}.'
                 )
 
             run_client = client.run(record.run_id)
@@ -127,7 +146,9 @@ async def find_or_start(
 
             if run is None or run.status == 'FAILED':
                 previous_run_ids = [*record.previous_run_ids, record.run_id]
-                run = await self._start(name, actor_id=actor_id, start_run=start_run, previous_run_ids=previous_run_ids)
+                run = await self._start(
+                    name, actor_id=actor_id, task_id=task_id, start_run=start_run, previous_run_ids=previous_run_ids
+                )
                 return run, True
 
             if run.status in _RESURRECTABLE_STATUSES:
@@ -160,12 +181,14 @@ async def _start(
         self,
         name: str,
         *,
-        actor_id: str,
+        actor_id: str | None,
+        task_id: str | None,
         start_run: Callable[[], Awaitable[Run]],
         previous_run_ids: list[str],
     ) -> Run:
         run = await start_run()
-        await self._save(name, ChildRunRecord(actor_id=actor_id, run_id=run.id, previous_run_ids=previous_run_ids))
+        record = ChildRunRecord(actor_id=actor_id, task_id=task_id, run_id=run.id, previous_run_ids=previous_run_ids)
+        await self._save(name, record)
         return run
 
     async def _load(self) -> dict[str, ChildRunRecord]:

tests/e2e/test_actor_child_runs.py

@@ -26,12 +26,12 @@ async def main() -> None:
             child_run_id = await Actor.get_value('child_run_id')
 
             if child_run_id is None:
-                run = await Actor.start(actor_id=actor_id, run_input={'is_child': True}, name='child')
+                run = await Actor.start(actor_id=actor_id, run_input={'is_child': True}, run_name='child')
                 await Actor.set_value('child_run_id', run.id)
                 await Actor.reboot()
                 return
 
-            run = await Actor.call(actor_id=actor_id, run_input={'is_child': True}, name='child')
+            run = await Actor.call(actor_id=actor_id, run_input={'is_child': True}, run_name='child')
             assert run is not None, 'run is None'
             assert run.id == child_run_id, f'run.id={run.id}, child_run_id={child_run_id}'
             assert run.status == 'SUCCEEDED', f'run.status={run.status}'
@@ -68,7 +68,7 @@ async def main() -> None:
             child_run_id = await Actor.get_value('child_run_id')
 
             if child_run_id is None:
-                run = await Actor.start(actor_id=actor_id, run_input={'is_child': True}, name='child')
+                run = await Actor.start(actor_id=actor_id, run_input={'is_child': True}, run_name='child')
                 await Actor.set_value('child_run_id', run.id)
                 run_client = Actor.apify_client.run(run.id)
                 await run_client.abort()
@@ -78,7 +78,7 @@ async def main() -> None:
                 await Actor.reboot()
                 return
 
-            run = await Actor.start(actor_id=actor_id, run_input={'is_child': True}, name='child')
+            run = await Actor.start(actor_id=actor_id, run_input={'is_child': True}, run_name='child')
             try:
                 assert run.id == child_run_id, f'run.id={run.id}, child_run_id={child_run_id}'
                 assert run.status in {'READY', 'RUNNING'}, f'run.status={run.status}'