diff --git a/docs/02_concepts/06_interacting_with_other_actors.mdx b/docs/02_concepts/06_interacting_with_other_actors.mdx index 1a362d7a1..2d7fef905 100644 --- a/docs/02_concepts/06_interacting_with_other_actors.mdx +++ b/docs/02_concepts/06_interacting_with_other_actors.mdx @@ -8,6 +8,7 @@ import RunnableCodeBlock from '@site/src/components/RunnableCodeBlock'; import InteractingStartExample from '!!raw-loader!roa-loader!./code/06_interacting_start.py'; import InteractingCallExample from '!!raw-loader!roa-loader!./code/06_interacting_call.py'; +import InteractingStartTaskExample from '!!raw-loader!roa-loader!./code/06_interacting_start_task.py'; import InteractingCallTaskExample from '!!raw-loader!roa-loader!./code/06_interacting_call_task.py'; import InteractingMetamorphExample from '!!raw-loader!roa-loader!./code/06_interacting_metamorph.py'; import InteractingAbortExample from '!!raw-loader!roa-loader!./code/06_interacting_abort.py'; @@ -34,6 +35,14 @@ The `Actor.call` method starts another {InteractingCallExample} +## Actor start task + +The `Actor.start_task` method starts an [Actor task](https://docs.apify.com/platform/actors/tasks) on the Apify platform, and immediately returns the details of the started Actor run. + + + {InteractingStartTaskExample} + + ## Actor call task The `Actor.call_task` method starts an [Actor task](https://docs.apify.com/platform/actors/tasks) on the Apify platform, and waits for the started Actor run to finish. diff --git a/docs/02_concepts/code/06_interacting_start_task.py b/docs/02_concepts/code/06_interacting_start_task.py new file mode 100644 index 000000000..43dbcb1ba --- /dev/null +++ b/docs/02_concepts/code/06_interacting_start_task.py @@ -0,0 +1,16 @@ +import asyncio + +from apify import Actor + + +async def main() -> None: + async with Actor: + # Start the Actor task by its ID without waiting for it to finish. + actor_run = await Actor.start_task(task_id='Z3m6FPSj0GYZ25rQc') + + # Log the run ID, which you can use to check on the run later. + Actor.log.info(f'Started task run: {actor_run.id}') + + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 317d7d9c8..1bb9857c1 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -1153,6 +1153,62 @@ async def call( return run + @_ensure_context + async def start_task( + self, + task_id: str, + task_input: dict | None = None, + *, + build: str | None = None, + max_items: int | None = None, + restart_on_error: bool | None = None, + memory_mbytes: int | None = None, + timeout: timedelta | Literal['inherit'] | None = None, + webhooks: list[Webhook] | None = None, + token: str | None = None, + ) -> Run: + """Start an Actor task on the Apify Platform. + + Unlike `Actor.call_task`, this method just starts the run without waiting for finish. To wait for the run to + finish, use `Actor.call_task` instead. + + Note that an Actor task is a saved input configuration and options for an Actor. If you want to run an Actor + directly rather than an Actor task, use `Actor.start`. + + Args: + task_id: The ID of the Actor task to be run. + task_input: Overrides the input to pass to the Actor run. + token: The Apify API token to use for this request (defaults to the `APIFY_TOKEN` environment variable). + build: Specifies the Actor build to run. It can be either a build tag or build number. By default, + the run uses the build specified in the default run configuration for the Actor (typically latest). + max_items: Maximum number of dataset items you are charged for, for pay-per-result Actors. It caps the + charge, not the output, so the run can return fewer or more items than this. + restart_on_error: If true, the Task run process will be restarted whenever it exits with + a non-zero status code. + memory_mbytes: Memory limit for the run, in megabytes. By default, the run uses a memory limit specified + in the default run configuration for the Actor. + timeout: Optional timeout for the run. By default, the run uses timeout specified in + the default run configuration for the Actor. Using `inherit` will set timeout of the other Actor to the + time remaining from this Actor timeout. + webhooks: Optional webhooks (https://docs.apify.com/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, you do not have to add it again here. + + Returns: + Info about the started Actor run. + """ + client = self.new_client(token=token) if token else self.apify_client + task_client = client.task(task_id) + return await task_client.start( + task_input=task_input, + build=build, + max_items=max_items, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=self._resolve_run_timeout(timeout), + webhooks=to_client_representations(webhooks), + ) + @_ensure_context async def call_task( self, diff --git a/tests/e2e/test_actor_api_helpers.py b/tests/e2e/test_actor_api_helpers.py index 980c38638..3949ba8e4 100644 --- a/tests/e2e/test_actor_api_helpers.py +++ b/tests/e2e/test_actor_api_helpers.py @@ -202,6 +202,65 @@ async def main_outer() -> None: assert inner_output_record['value'] == f'{test_value}_XXX_{test_value}' +async def test_actor_starts_task( + make_actor: MakeActorFunction, + run_actor: RunActorFunction, + apify_client_async: ApifyClientAsync, +) -> None: + """`Actor.start_task` returns the run before it finishes, and the started task run completes with its output.""" + + async def main_inner() -> None: + async with Actor: + await asyncio.sleep(5) + actor_input = await Actor.get_input() or {} + test_value = actor_input.get('test_value') + await Actor.set_value('OUTPUT', f'{test_value}_XXX_{test_value}') + + async def main_outer() -> None: + async with Actor: + actor_input = await Actor.get_input() or {} + inner_task_id = actor_input.get('inner_task_id') + + assert inner_task_id is not None + + inner_run = await Actor.start_task(inner_task_id) + + assert inner_run.actor_task_id == inner_task_id + assert inner_run.status in {'READY', 'RUNNING'} + + inner_actor = await make_actor(label='start-task-inner', main_func=main_inner) + outer_actor = await make_actor(label='start-task-outer', main_func=main_outer) + + inner_actor_get_result = await inner_actor.get() + assert inner_actor_get_result is not None, 'Failed to get inner actor ID' + + inner_actor_id = inner_actor_get_result.id + test_value = crypto_random_object_id() + + task = await apify_client_async.tasks().create( + actor_id=inner_actor_id, + name=generate_unique_resource_name('actor-start-task'), + task_input={'test_value': test_value}, + ) + + try: + run_result_outer = await run_actor( + outer_actor, + run_input={'inner_task_id': task.id}, + force_permission_level='FULL_PERMISSIONS', + ) + + assert run_result_outer.status == 'SUCCEEDED' + + await inner_actor.last_run().wait_for_finish(wait_duration=timedelta(seconds=600)) + + inner_output_record = await inner_actor.last_run().key_value_store().get_record('OUTPUT') + assert inner_output_record is not None + assert inner_output_record['value'] == f'{test_value}_XXX_{test_value}' + finally: + await apify_client_async.task(task.id).delete() + + async def test_actor_calls_task( make_actor: MakeActorFunction, run_actor: RunActorFunction, diff --git a/tests/unit/actor/test_actor_helpers.py b/tests/unit/actor/test_actor_helpers.py index c36228dab..a5b678358 100644 --- a/tests/unit/actor/test_actor_helpers.py +++ b/tests/unit/actor/test_actor_helpers.py @@ -107,6 +107,21 @@ async def test_call_actor_task(apify_client_async_patcher: ApifyClientAsyncPatch assert apify_client_async_patcher.calls['task']['call'][0][0][0].resource_id == task_id +async def test_start_actor_task(apify_client_async_patcher: ApifyClientAsyncPatcher, fake_actor_run: Run) -> None: + """`Actor.start_task` starts the task through the client's `task.start` and returns the run.""" + apify_client_async_patcher.patch('task', 'start', return_value=fake_actor_run) + task_id = 'some-task-id' + + async with Actor: + run = await Actor.start_task(task_id, {'foo': 'bar'}) + + assert run is fake_actor_run + calls = apify_client_async_patcher.calls['task']['start'] + assert len(calls) == 1 + assert calls[0][0][0].resource_id == task_id + assert calls[0][1]['task_input'] == {'foo': 'bar'} + + async def test_start_actor(apify_client_async_patcher: ApifyClientAsyncPatcher, fake_actor_run: Run) -> None: apify_client_async_patcher.patch('actor', 'start', return_value=fake_actor_run) actor_id = 'some-id' @@ -123,6 +138,7 @@ async def test_start_actor(apify_client_async_patcher: ApifyClientAsyncPatcher, [ pytest.param('actor', 'start', 'start', id='start'), pytest.param('actor', 'call', 'call', id='call'), + pytest.param('task', 'start', 'start_task', id='start_task'), pytest.param('task', 'call', 'call_task', id='call_task'), ], ) @@ -133,7 +149,7 @@ async def test_max_items_forwarded_to_client( client_method: str, sdk_method: str, ) -> None: - """`max_items` passed to `Actor.start`, `Actor.call` or `Actor.call_task` reaches the API client.""" + """`max_items` passed to any of the run-starting helpers reaches the API client.""" apify_client_async_patcher.patch(client_resource, client_method, return_value=fake_actor_run) async with Actor: @@ -300,6 +316,7 @@ async def listener(_data: object) -> None: _ACTOR_REMOTE_METHODS = [ pytest.param('actor', 'start', 'start', 'some-actor-id', id='start'), pytest.param('actor', 'call', 'call', 'some-actor-id', id='call'), + pytest.param('task', 'start', 'start_task', 'some-task-id', id='start_task'), pytest.param('task', 'call', 'call_task', 'some-task-id', id='call_task'), ] @@ -314,7 +331,7 @@ async def test_remote_method_with_webhooks( actor_method_name: str, entity_id: str, ) -> None: - """Test that start/call/call_task correctly serialize webhooks.""" + """Test that start/call/start_task/call_task correctly serialize webhooks.""" apify_client_async_patcher.patch(client_resource, client_method, return_value=fake_actor_run) async with Actor: @@ -341,7 +358,7 @@ async def test_remote_method_with_timedelta_timeout( actor_method_name: str, entity_id: str, ) -> None: - """Test that start/call/call_task accept a timedelta timeout.""" + """Test that start/call/start_task/call_task accept a timedelta timeout.""" apify_client_async_patcher.patch(client_resource, client_method, return_value=fake_actor_run) async with Actor: @@ -364,7 +381,7 @@ async def test_remote_method_with_invalid_timeout( actor_method_name: str, entity_id: str, ) -> None: - """Test that start/call/call_task raise ValueError for invalid timeout.""" + """Test that start/call/start_task/call_task raise ValueError for invalid timeout.""" apify_client_async_patcher.patch(client_resource, client_method, return_value=fake_actor_run) async with Actor: @@ -437,19 +454,28 @@ async def test_actor_start_and_call_skipped_when_no_inherited_time_remains( assert apify_client_async_patcher.calls['actor'][method_name][0][1]['run_timeout'] == timedelta(seconds=1) -async def test_actor_call_task_skipped_when_no_inherited_time_remains( +@pytest.mark.parametrize( + ('client_method', 'actor_method_name'), + [ + pytest.param('start', 'start_task', id='start_task'), + pytest.param('call', 'call_task', id='call_task'), + ], +) +async def test_actor_task_methods_skipped_when_no_inherited_time_remains( apify_client_async_patcher: ApifyClientAsyncPatcher, + client_method: str, + actor_method_name: str, ) -> None: - """Test that Actor.call_task with `timeout='inherit'` is skipped when the run is past its timeout.""" - apify_client_async_patcher.patch('task', 'call', return_value=Mock()) + """Test that start_task/call_task with `timeout='inherit'` is skipped when the run is past its timeout.""" + apify_client_async_patcher.patch('task', client_method, return_value=Mock()) async with Actor: Actor.configuration.is_at_home = True Actor.configuration.timeout_at = datetime.now(tz=UTC) - timedelta(minutes=5) - await Actor.call_task('some-task-id', timeout='inherit') + await getattr(Actor, actor_method_name)('some-task-id', timeout='inherit') - assert len(apify_client_async_patcher.calls['task']['call']) == 1 - assert apify_client_async_patcher.calls['task']['call'][0][1]['run_timeout'] == timedelta(seconds=1) + assert len(apify_client_async_patcher.calls['task'][client_method]) == 1 + assert apify_client_async_patcher.calls['task'][client_method][0][1]['run_timeout'] == timedelta(seconds=1) async def test_reboot_runs_all_listeners_even_when_one_fails(