feat: Add StatusMessageRedirector

src/apify_client/clients/resource_clients/actor.py

@@ -317,7 +317,8 @@ def call(
                 waits indefinitely.
             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.
+                will redirect logs to the provided logger. The logger is also used to capture status and status message
+                of the other Actor run.
 
         Returns:
             The run object.
@@ -336,12 +337,11 @@ def call(
             return self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
         run_client = self.root_client.run(run_id=started_run['id'])
+
         if logger == 'default':
-            log_context = run_client.get_streamed_log()
-        else:
-            log_context = run_client.get_streamed_log(to_logger=logger)
+            logger = None
 
-        with log_context:
+        with run_client.get_status_message_redirector(to_logger=logger), run_client.get_streamed_log(to_logger=logger):
             return self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
     def build(
@@ -722,7 +722,8 @@ async def call(
                 waits indefinitely.
             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.
+                will redirect logs to the provided logger. The logger is also used to capture status and status message
+                of the other Actor run.
 
         Returns:
             The run object.
@@ -742,12 +743,14 @@ async def call(
             return await self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
         run_client = self.root_client.run(run_id=started_run['id'])
+
         if logger == 'default':
-            log_context = await run_client.get_streamed_log()
-        else:
-            log_context = await run_client.get_streamed_log(to_logger=logger)
+            logger = None
+
+        status_redirector = await run_client.get_status_message_redirector(to_logger=logger)
+        streamed_log = await run_client.get_streamed_log(to_logger=logger)
 
-        async with log_context:
+        async with status_redirector, streamed_log:
             return await self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
     async def build(

src/apify_client/clients/resource_clients/log.py

@@ -4,13 +4,15 @@
 import logging
 import re
 import threading
+import time
 from asyncio import Task
 from contextlib import asynccontextmanager, contextmanager
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from threading import Thread
 from typing import TYPE_CHECKING, Any, cast
 
 from apify_shared.utils import ignore_docs
+from typing_extensions import Self
 
 from apify_client._errors import ApifyApiError
 from apify_client._utils import catch_not_found_or_throw
@@ -23,6 +25,8 @@
     import httpx
     from typing_extensions import Self
 
+    from apify_client.clients import RunClient, RunClientAsync
+
 
 class LogClient(ResourceClient):
     """Sub-client for manipulating logs."""
@@ -378,3 +382,159 @@ async def _stream_log(self) -> None:
 
             # If the stream is finished, then the last part will be also processed.
             self._log_buffer_content(include_last_part=True)
+
+
+class StatusMessageRedirector:
+    """Utility class for logging status messages from another Actor run.
+
+    Status message is logged at fixed time intervals, and there is no guarantee that all messages will be logged,
+    especially in cases of frequent status message changes.
+    """
+
+    _force_propagate = False
+    # This is final sleep time to try to get the last status and status message of finished Actor run.
+    # The status and status message can get set on the Actor run with a delay. Sleep time does not guarantee that the
+    # final message will be captured, but increases the chances of that.
+    _final_sleep_time_s = 6
+
+    def __init__(self, *, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=5)) -> None:
+        """Initialize `StatusMessageRedirector`.
+
+        Args:
+            to_logger: The logger to which the status message will be redirected.
+            check_period: The period with which the status message will be polled.
+        """
+        self._to_logger = to_logger
+        self._to_logger.propagate = self._force_propagate
+        self._check_period = check_period.total_seconds()
+        self._last_status_message = ''
+
+    def _log_run_data(self, run_data: dict[str, Any] | None) -> bool:
+        """Get relevant run data, log them if changed and return `True` if more data is expected.
+
+        Args:
+            run_data: The dictionary that contains the run data.
+
+        Returns:
+              `True` if more data is expected, `False` otherwise.
+        """
+        if run_data is not None:
+            status = run_data.get('status', 'Unknown status')
+            status_message = run_data.get('statusMessage', '')
+            new_status_message = f'Status: {status}, Message: {status_message}'
+
+            if new_status_message != self._last_status_message:
+                self._last_status_message = new_status_message
+                self._to_logger.info(new_status_message)
+
+            return not (run_data.get('isStatusMessageTerminal', False))
+        return True
+
+
+class StatusMessageRedirectorAsync(StatusMessageRedirector):
+    """Async variant of `StatusMessageRedirector` that is logging in task."""
+
+    def __init__(
+        self, *, run_client: RunClientAsync, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=1)
+    ) -> None:
+        """Initialize `StatusMessageRedirectorAsync`.
+
+        Args:
+            run_client: The client for run that will be used to get a status and message.
+            to_logger: The logger to which the status message will be redirected.
+            check_period: The period with which the status message will be polled.
+        """
+        super().__init__(to_logger=to_logger, check_period=check_period)
+        self._run_client = run_client
+        self._logging_task: Task | None = None
+
+    def start(self) -> Task:
+        """Start the logging task. The caller has to handle any cleanup by manually calling the `stop` method."""
+        if self._logging_task:
+            raise RuntimeError('Logging task already active')
+        self._logging_task = asyncio.create_task(self._log_changed_status_message())
+        return self._logging_task
+
+    def stop(self) -> None:
+        """Stop the logging task."""
+        if not self._logging_task:
+            raise RuntimeError('Logging task is not active')
+
+        self._logging_task.cancel()
+        self._logging_task = None
+
+    async def __aenter__(self) -> Self:
+        """Start the logging task within the context. Exiting the context will cancel the logging task."""
+        self.start()
+        return self
+
+    async def __aexit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> None:
+        """Cancel the logging task."""
+        await asyncio.sleep(self._final_sleep_time_s)
+        self.stop()
+
+    async def _log_changed_status_message(self) -> None:
+        while True:
+            run_data = await self._run_client.get()
+            if not self._log_run_data(run_data):
+                break
+            await asyncio.sleep(self._check_period)
+
+
+class StatusMessageRedirectorSync(StatusMessageRedirector):
+    """Sync variant of `StatusMessageRedirector` that is logging in thread."""
+
+    def __init__(
+        self, *, run_client: RunClient, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=1)
+    ) -> None:
+        """Initialize `StatusMessageRedirectorSync`.
+
+        Args:
+            run_client: The client for run that will be used to get a status and message.
+            to_logger: The logger to which the status message will be redirected.
+            check_period: The period with which the status message will be polled.
+        """
+        super().__init__(to_logger=to_logger, check_period=check_period)
+        self._run_client = run_client
+        self._logging_thread: Thread | None = None
+        self._stop_logging = False
+
+    def start(self) -> Thread:
+        """Start the logging thread. The caller has to handle any cleanup by manually calling the `stop` method."""
+        if self._logging_thread:
+            raise RuntimeError('Logging thread already active')
+        self._stop_logging = False
+        self._logging_thread = threading.Thread(target=self._log_changed_status_message)
+        self._logging_thread.start()
+        return self._logging_thread
+
+    def stop(self) -> None:
+        """Signal the _logging_thread thread to stop logging and wait for it to finish."""
+        if not self._logging_thread:
+            raise RuntimeError('Logging thread is not active')
+        time.sleep(self._final_sleep_time_s)
+        self._stop_logging = True
+        self._logging_thread.join()
+        self._logging_thread = None
+        self._stop_logging = False
+
+    def __enter__(self) -> Self:
+        """Start the logging task within the context. Exiting the context will cancel the logging task."""
+        self.start()
+        return self
+
+    def __exit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> None:
+        """Cancel the logging task."""
+        self.stop()
+
+    def _log_changed_status_message(self) -> None:
+        while True:
+            if not self._log_run_data(self._run_client.get()):
+                break
+            if self._stop_logging:
+                break
+            time.sleep(self._check_period)

src/apify_client/clients/resource_clients/run.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
 import json
+import logging
 import random
 import string
 import time
+from datetime import timedelta
 from typing import TYPE_CHECKING, Any
 
 from apify_shared.utils import filter_out_none_values_recursively, ignore_docs, parse_date_fields
@@ -16,6 +18,8 @@
 from apify_client.clients.resource_clients.log import (
     LogClient,
     LogClientAsync,
+    StatusMessageRedirectorAsync,
+    StatusMessageRedirectorSync,
     StreamedLogAsync,
     StreamedLogSync,
 )
@@ -318,6 +322,34 @@ def charge(
             ),
         )
 
+    def get_status_message_redirector(
+        self, to_logger: logging.Logger | None = None, check_period: timedelta = timedelta(seconds=1)
+    ) -> StatusMessageRedirectorSync:
+        """Get `StatusMessageRedirector` instance that can be used to redirect logs.
+
+        `StatusMessageRedirector` can be directly called or used as a context manager.
+
+        Args:
+            to_logger: `Logger` used for logging the status and status messages. If not provided, a new logger is
+            created.
+            check_period: The period with which the status message will be polled.
+
+        Returns:
+            `StatusMessageRedirector` instance for redirected logs.
+        """
+        run_data = self.get()
+        run_id = run_data.get('id', '') if run_data else ''
+
+        actor_id = run_data.get('actId', '') if run_data else ''
+        actor_data = self.root_client.actor(actor_id=actor_id).get() or {}
+        actor_name = actor_data.get('name', '') if run_data else ''
+
+        if not to_logger:
+            name = '-'.join(part for part in (actor_name, run_id) if part)
+            to_logger = create_redirect_logger(f'apify.{name}')
+
+        return StatusMessageRedirectorSync(run_client=self, to_logger=to_logger, check_period=check_period)
+
 
 class RunClientAsync(ActorJobBaseClientAsync):
     """Async sub-client for manipulating a single Actor run."""
@@ -612,3 +644,33 @@ async def charge(
                 }
             ),
         )
+
+    async def get_status_message_redirector(
+        self,
+        to_logger: logging.Logger | None = None,
+        check_period: timedelta = timedelta(seconds=1),
+    ) -> StatusMessageRedirectorAsync:
+        """Get `StatusMessageRedirector` instance that can be used to redirect logs.
+
+        `StatusMessageRedirector` can be directly called or used as a context manager.
+
+        Args:
+            to_logger: `Logger` used for logging the status and status messages. If not provided, a new logger is
+            created.
+            check_period: The period with which the status message will be polled.
+
+        Returns:
+            `StatusMessageRedirector` instance for redirected logs.
+        """
+        run_data = await self.get()
+        run_id = run_data.get('id', '') if run_data else ''
+
+        actor_id = run_data.get('actId', '') if run_data else ''
+        actor_data = await self.root_client.actor(actor_id=actor_id).get() or {}
+        actor_name = actor_data.get('name', '') if run_data else ''
+
+        if not to_logger:
+            name = '-'.join(part for part in (actor_name, run_id) if part)
+            to_logger = create_redirect_logger(f'apify.{name}')
+
+        return StatusMessageRedirectorAsync(run_client=self, to_logger=to_logger, check_period=check_period)