Review comments

Author: Pijukatel

src/apify_client/clients/resource_clients/actor.py

@@ -341,7 +341,7 @@ def call(
         if logger == 'default':
             logger = None
 
-        with run_client.get_status_message_redirector(to_logger=logger), run_client.get_streamed_log(to_logger=logger):
+        with run_client.get_status_message_watcher(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(
@@ -747,7 +747,7 @@ async def call(
         if logger == 'default':
             logger = None
 
-        status_redirector = await run_client.get_status_message_redirector(to_logger=logger)
+        status_redirector = await run_client.get_status_message_watcher(to_logger=logger)
         streamed_log = await run_client.get_streamed_log(to_logger=logger)
 
         async with status_redirector, streamed_log:

src/apify_client/clients/resource_clients/log.py

@@ -232,9 +232,9 @@ def __init__(self, to_logger: logging.Logger, *, from_start: bool = True) -> Non
                 logs for long-running actors in stand-by.
 
         """
-        self._to_logger = to_logger
         if self._force_propagate:
             to_logger.propagate = True
+        self._to_logger = to_logger
         self._stream_buffer = list[bytes]()
         self._split_marker = re.compile(rb'(?:\n|^)(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z)')
         self._relevancy_time_limit: datetime | None = None if from_start else datetime.now(tz=timezone.utc)
@@ -384,7 +384,7 @@ async def _stream_log(self) -> None:
             self._log_buffer_content(include_last_part=True)
 
 
-class StatusMessageRedirector:
+class StatusMessageWatcher:
     """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,
@@ -398,12 +398,14 @@ class StatusMessageRedirector:
     _final_sleep_time_s = 6
 
     def __init__(self, *, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=5)) -> None:
-        """Initialize `StatusMessageRedirector`.
+        """Initialize `StatusMessageWatcher`.
 
         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.
         """
+        if self._force_propagate:
+            to_logger.propagate = True
         self._to_logger = to_logger
         self._to_logger.propagate = self._force_propagate
         self._check_period = check_period.total_seconds()
@@ -431,13 +433,13 @@ def _log_run_data(self, run_data: dict[str, Any] | None) -> bool:
         return True
 
 
-class StatusMessageRedirectorAsync(StatusMessageRedirector):
-    """Async variant of `StatusMessageRedirector` that is logging in task."""
+class StatusMessageWatcherAsync(StatusMessageWatcher):
+    """Async variant of `StatusMessageWatcher` that is logging in task."""
 
     def __init__(
         self, *, run_client: RunClientAsync, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=1)
     ) -> None:
-        """Initialize `StatusMessageRedirectorAsync`.
+        """Initialize `StatusMessageWatcherAsync`.
 
         Args:
             run_client: The client for run that will be used to get a status and message.
@@ -483,13 +485,13 @@ async def _log_changed_status_message(self) -> None:
             await asyncio.sleep(self._check_period)
 
 
-class StatusMessageRedirectorSync(StatusMessageRedirector):
-    """Sync variant of `StatusMessageRedirector` that is logging in thread."""
+class StatusMessageWatcherSync(StatusMessageWatcher):
+    """Sync variant of `StatusMessageWatcher` that is logging in thread."""
 
     def __init__(
         self, *, run_client: RunClient, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=1)
     ) -> None:
-        """Initialize `StatusMessageRedirectorSync`.
+        """Initialize `StatusMessageWatcherSync`.
 
         Args:
             run_client: The client for run that will be used to get a status and message.

src/apify_client/clients/resource_clients/run.py

@@ -18,8 +18,8 @@
 from apify_client.clients.resource_clients.log import (
     LogClient,
     LogClientAsync,
-    StatusMessageRedirectorAsync,
-    StatusMessageRedirectorSync,
+    StatusMessageWatcherAsync,
+    StatusMessageWatcherSync,
     StreamedLogAsync,
     StreamedLogSync,
 )
@@ -274,7 +274,7 @@ def get_streamed_log(self, to_logger: logging.Logger | None = None, *, from_star
             `StreamedLog` instance for redirected logs.
         """
         run_data = self.get()
-        run_id = run_data.get('id', '') if run_data else ''
+        run_id = f'runId:{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 {}
@@ -322,23 +322,23 @@ def charge(
             ),
         )
 
-    def get_status_message_redirector(
+    def get_status_message_watcher(
         self, to_logger: logging.Logger | None = None, check_period: timedelta = timedelta(seconds=1)
-    ) -> StatusMessageRedirectorSync:
-        """Get `StatusMessageRedirector` instance that can be used to redirect logs.
+    ) -> StatusMessageWatcherSync:
+        """Get `StatusMessageWatcher` instance that can be used to redirect status and status messages to logs.
 
-        `StatusMessageRedirector` can be directly called or used as a context manager.
+        `StatusMessageWatcher` 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.
+            `StatusMessageWatcher` instance.
         """
         run_data = self.get()
-        run_id = run_data.get('id', '') if run_data else ''
+        run_id = f'runId:{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 {}
@@ -348,7 +348,7 @@ def get_status_message_redirector(
             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)
+        return StatusMessageWatcherSync(run_client=self, to_logger=to_logger, check_period=check_period)
 
 
 class RunClientAsync(ActorJobBaseClientAsync):
@@ -598,7 +598,7 @@ async def get_streamed_log(
             `StreamedLog` instance for redirected logs.
         """
         run_data = await self.get()
-        run_id = run_data.get('id', '') if run_data else ''
+        run_id = f'runId:{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 {}
@@ -645,25 +645,25 @@ async def charge(
             ),
         )
 
-    async def get_status_message_redirector(
+    async def get_status_message_watcher(
         self,
         to_logger: logging.Logger | None = None,
         check_period: timedelta = timedelta(seconds=1),
-    ) -> StatusMessageRedirectorAsync:
-        """Get `StatusMessageRedirector` instance that can be used to redirect logs.
+    ) -> StatusMessageWatcherAsync:
+        """Get `StatusMessageWatcher` instance that can be used to redirect status and status messages to logs.
 
-        `StatusMessageRedirector` can be directly called or used as a context manager.
+        `StatusMessageWatcher` 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.
+            `StatusMessageWatcher` instance.
         """
         run_data = await self.get()
-        run_id = run_data.get('id', '') if run_data else ''
+        run_id = f'runId:{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 {}
@@ -673,4 +673,4 @@ async def get_status_message_redirector(
             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)
+        return StatusMessageWatcherAsync(run_client=self, to_logger=to_logger, check_period=check_period)

tests/unit/test_logging.py

@@ -14,7 +14,7 @@
 
 from apify_client import ApifyClient, ApifyClientAsync
 from apify_client._logging import RedirectLogFormatter
-from apify_client.clients.resource_clients.log import StatusMessageRedirector, StreamedLog
+from apify_client.clients.resource_clients.log import StatusMessageWatcher, StreamedLog
 
 _MOCKED_API_URL = 'https://example.com'
 _MOCKED_RUN_ID = 'mocked_run_id'
@@ -165,17 +165,17 @@ def close(self) -> None:
 def propagate_stream_logs() -> None:
     # Enable propagation of logs to the caplog fixture
     StreamedLog._force_propagate = True
-    StatusMessageRedirector._force_propagate = True
+    StatusMessageWatcher._force_propagate = True
     logging.getLogger(f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}').setLevel(logging.DEBUG)
 
 
 @pytest.fixture
 def reduce_final_timeout_for_status_message_redirector() -> None:
-    """Reduce timeout used by the `StatusMessageRedirector`
+    """Reduce timeout used by the `StatusMessageWatcher`
 
     This timeout makes sense on the platform, but in tests it is better to reduce it to speed up the tests.
     """
-    StatusMessageRedirector._final_sleep_time_s = 2
+    StatusMessageWatcher._final_sleep_time_s = 2
 
 
 @pytest.mark.parametrize(
@@ -265,7 +265,7 @@ async def test_actor_call_redirect_logs_to_default_logger_async(
     """Test that logs are redirected correctly to the default logger.
 
     Caplog contains logs before formatting, so formatting is not included in the test expectations."""
-    logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
+    logger_name = f'apify.{_MOCKED_ACTOR_NAME}-runId:{_MOCKED_RUN_ID}'
     logger = logging.getLogger(logger_name)
     actor_client = ApifyClientAsync(token='mocked_token', api_url=_MOCKED_API_URL).actor(actor_id=_MOCKED_ACTOR_ID)
 
@@ -293,7 +293,7 @@ def test_actor_call_redirect_logs_to_default_logger_sync(
     """Test that logs are redirected correctly to the default logger.
 
     Caplog contains logs before formatting, so formatting is not included in the test expectations."""
-    logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
+    logger_name = f'apify.{_MOCKED_ACTOR_NAME}-runId:{_MOCKED_RUN_ID}'
     logger = logging.getLogger(logger_name)
     actor_client = ApifyClient(token='mocked_token', api_url=_MOCKED_API_URL).actor(actor_id=_MOCKED_ACTOR_ID)
 
@@ -397,7 +397,7 @@ async def test_redirect_status_message_async(
 
     logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
 
-    status_message_redirector = await run_client.get_status_message_redirector(check_period=timedelta(seconds=0))
+    status_message_redirector = await run_client.get_status_message_watcher(check_period=timedelta(seconds=0))
     with caplog.at_level(logging.DEBUG, logger=logger_name):
         async with status_message_redirector:
             # Do stuff while the status from the other Actor is being redirected to the logs.
@@ -422,7 +422,7 @@ def test_redirect_status_message_sync(
 
     logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
 
-    status_message_redirector = run_client.get_status_message_redirector(check_period=timedelta(seconds=0))
+    status_message_redirector = run_client.get_status_message_watcher(check_period=timedelta(seconds=0))
     with caplog.at_level(logging.DEBUG, logger=logger_name), status_message_redirector:
         # Do stuff while the status from the other Actor is being redirected to the logs.
         time.sleep(3)