Merge branch 'feat/child-runs-api' into feat/abort-child-runs-with-parent

Author: vdusek

src/apify/_actor.py

@@ -951,7 +951,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,
         abort_with_parent: bool = False,
     ) -> Run:
         """Run an Actor on the Apify platform.
@@ -979,21 +979,21 @@ 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`.
             abort_with_parent: If true, the child run is gracefully aborted when this Actor run is gracefully
-                aborted. It requires `name`, and the value is recorded under it, replacing the one from an earlier
+                aborted. It requires `run_name`, and the value is recorded under it, replacing the one from an earlier
                 call. A hard abort, a timeout or a crash of this Actor run leaves the child running.
 
         Returns:
             Info about the started Actor run
         """
-        if abort_with_parent and name is None:
-            raise ValueError('`abort_with_parent` requires `name`, since only named child runs are tracked.')
+        if abort_with_parent and run_name is None:
+            raise ValueError('`abort_with_parent` requires `run_name`, since only named child runs are tracked.')
 
         client = self.new_client(token=token) if token else self.apify_client
 
@@ -1020,11 +1020,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,
@@ -1091,7 +1091,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,
         abort_with_parent: bool = False,
     ) -> Run:
         """Start an Actor on the Apify Platform and wait for it to finish before returning.
@@ -1122,21 +1122,21 @@ 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`.
             abort_with_parent: If true, the child run is gracefully aborted when this Actor run is gracefully
-                aborted. It requires `name`, and the value is recorded under it, replacing the one from an earlier
+                aborted. It requires `run_name`, and the value is recorded under it, replacing the one from an earlier
                 call. A hard abort, a timeout or a crash of this Actor run leaves the child running.
 
         Returns:
             Info about the started Actor run.
         """
-        if abort_with_parent and name is None:
-            raise ValueError('`abort_with_parent` requires `name`, since only named child runs are tracked.')
+        if abort_with_parent and run_name is None:
+            raise ValueError('`abort_with_parent` requires `run_name`, since only named child runs are tracked.')
 
         client = self.new_client(token=token) if token else self.apify_client
 
@@ -1151,7 +1151,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,
@@ -1167,7 +1167,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(
@@ -1203,7 +1203,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,
@@ -1216,6 +1217,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(
@@ -1261,9 +1263,9 @@ async def _wait_for_child_run(
     async def child_runs(self) -> dict[str, ChildRunInfo]:
         """Get the named child runs of this Actor run, with their current state.
 
-        Every run started by `Actor.start` or `Actor.call` with a `name` is included, even one started before a
-        migration or resurrection of this Actor run. Runs started without a `name` are not tracked. Each run is fetched
-        from the API when this method is called, so the result is a snapshot.
+        Every run started by `Actor.start`, `Actor.call` or `Actor.call_task` with a `run_name` is included, even one
+        started before a migration or resurrection of this Actor run. Runs started without a `run_name` are not
+        tracked. Each run is fetched from the API when this method is called, so the result is a snapshot.
 
         Returns:
             The child runs by name.
@@ -1283,6 +1285,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.
 
@@ -1308,6 +1311,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.
@@ -1324,15 +1333,41 @@ 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,
+                abort_with_parent=False,
+            )
+            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
@@ -38,8 +38,11 @@ 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."""
@@ -50,14 +53,27 @@ class ChildRunRecord(BaseModel):
     abort_with_parent: bool = False
     """Whether the current run is aborted when this Actor run is gracefully aborted."""
 
+    @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)
 class ChildRunInfo:
     """A named child run of this Actor run, as returned by `Actor.child_runs`."""
 
-    actor_id: str
-    """The Actor ID or name the child was started with, as the caller passed it."""
+    actor_id: str | 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
+    """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."""
@@ -95,7 +111,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]],
@@ -110,8 +127,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, resurrect and abort 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.
             abort_with_parent: Whether to abort the run when this Actor run is gracefully aborted. It replaces
                 the value recorded under `name`.
@@ -123,10 +141,11 @@ async def find_or_start(
             records = await self._load()
             record = records.get(name)
 
-            if record is not None and record.actor_id != actor_id:
+            if record is not None and (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)}.'
                 )
 
             self._clients[name] = client
@@ -135,6 +154,7 @@ async def find_or_start(
                 run = await self._start(
                     name,
                     actor_id=actor_id,
+                    task_id=task_id,
                     start_run=start_run,
                     previous_run_ids=[],
                     abort_with_parent=abort_with_parent,
@@ -151,6 +171,7 @@ async def find_or_start(
                 run = await self._start(
                     name,
                     actor_id=actor_id,
+                    task_id=task_id,
                     start_run=start_run,
                     previous_run_ids=[*record.previous_run_ids, record.run_id],
                     abort_with_parent=abort_with_parent,
@@ -179,6 +200,7 @@ async def list_runs(self, client: ApifyClientAsync) -> dict[str, ChildRunInfo]:
         return {
             name: ChildRunInfo(
                 actor_id=record.actor_id,
+                task_id=record.task_id,
                 run_id=record.run_id,
                 run=run,
                 previous_run_ids=list(record.previous_run_ids),
@@ -219,14 +241,16 @@ 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],
         abort_with_parent: bool,
     ) -> Run:
         run = await start_run()
         record = ChildRunRecord(
             actor_id=actor_id,
+            task_id=task_id,
             run_id=run.id,
             previous_run_ids=previous_run_ids,
             abort_with_parent=abort_with_parent,

tests/e2e/test_actor_child_runs.py

@@ -30,12 +30,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}'
@@ -72,7 +72,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()
@@ -82,7 +82,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}'