Add from_start argument for streaming from stand-by actors

Update default logger to not duplicate handlers if already exists

Author: Pijukatel

src/apify_client/_logging.py

@@ -137,8 +137,14 @@ def create_redirect_logger(
         The created logger.
     """
     to_logger = logging.getLogger(name)
-
     to_logger.propagate = False
+
+    # Remove filters and handlers in case this logger already exists and was set up in some way.
+    for handler in to_logger.handlers:
+        to_logger.removeHandler(handler)
+    for log_filter in to_logger.filters:
+        to_logger.removeFilter(log_filter)
+
     handler = logging.StreamHandler()
     handler.setFormatter(RedirectLogFormatter())
     to_logger.addHandler(handler)

src/apify_client/clients/resource_clients/log.py

@@ -214,12 +214,24 @@ class StreamedLog:
     # Test related flag to enable propagation of logs to the `caplog` fixture during tests.
     _force_propagate = False
 
-    def __init__(self, to_logger: logging.Logger) -> None:
+    def __init__(self, to_logger: logging.Logger, *, from_start: bool = True) -> None:
+        """Initialize `StreamedLog`.
+
+        Args:
+            to_logger: The logger to which the logs will be redirected
+            from_start: If `True`, all logs from the start of the actor run will be redirected. If `False`, only newly
+                arrived logs will be redirected. This can be useful for redirecting only a small portion of relevant
+                logs for long-running actors in stand-by.
+
+        Returns:
+            The created logger.
+        """
         self._to_logger = to_logger
         if self._force_propagate:
             to_logger.propagate = True
         self._stream_buffer = list[str]()
         self._split_marker = re.compile(r'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z)')  # Ex:2025-05-12T15:35:59.429Z
+        self._from_start = from_start
 
     def _process_new_data(self, data: bytes) -> None:
         new_chunk = data.decode('utf-8')
@@ -265,8 +277,8 @@ def _guess_log_level_from_message(message: str) -> int:
 class StreamedLogSync(StreamedLog):
     """Sync variant of `StreamedLog` that is logging in threads."""
 
-    def __init__(self, log_client: LogClient, to_logger: logging.Logger) -> None:
-        super().__init__(to_logger=to_logger)
+    def __init__(self, log_client: LogClient, *, to_logger: logging.Logger, from_start: bool = True) -> None:
+        super().__init__(to_logger=to_logger, from_start=from_start)
         self._log_client = log_client
         self._streaming_thread: Thread | None = None
         self._stop_logging = False
@@ -302,7 +314,12 @@ def _stream_log(self) -> None:
         with self._log_client.stream(raw=True) as log_stream:
             if not log_stream:
                 return
+            # The first chunk contains all older logs from the start of the actor run until now.
+            skip_first_chunk = not self._from_start
             for data in log_stream.iter_bytes():
+                if skip_first_chunk:
+                    skip_first_chunk = False
+                    continue
                 self._process_new_data(data)
                 if self._stop_logging:
                     break
@@ -315,8 +332,8 @@ def _stream_log(self) -> None:
 class StreamedLogAsync(StreamedLog):
     """Async variant of `StreamedLog` that is logging in tasks."""
 
-    def __init__(self, log_client: LogClientAsync, to_logger: logging.Logger) -> None:
-        super().__init__(to_logger=to_logger)
+    def __init__(self, log_client: LogClientAsync, *, to_logger: logging.Logger, from_start: bool = True) -> None:
+        super().__init__(to_logger=to_logger, from_start=from_start)
         self._log_client = log_client
         self._streaming_task: Task | None = None
 
@@ -346,7 +363,12 @@ async def _stream_log(self) -> None:
         async with self._log_client.stream(raw=True) as log_stream:
             if not log_stream:
                 return
+            # The first chunk contains all older logs from the start of the actor run until now.
+            skip_first_chunk = not self._from_start
             async for data in log_stream.aiter_bytes():
+                if skip_first_chunk:
+                    skip_first_chunk = False
+                    continue
                 self._process_new_data(data)
 
             # If the stream is finished, then the last part will be also processed.

src/apify_client/clients/resource_clients/run.py

@@ -255,14 +255,19 @@ def log(self) -> LogClient:
             **self._sub_resource_init_options(resource_path='log'),
         )
 
-    def get_streamed_log(self, to_logger: logging.Logger | None = None, actor_name: str = '') -> StreamedLogSync:
+    def get_streamed_log(
+        self, to_logger: logging.Logger | None = None, *, actor_name: str = '', from_start: bool = True
+    ) -> StreamedLogSync:
         """Get `StreamedLog` instance that can be used to redirect logs.
 
          `StreamedLog` can be directly called or used as a context manager.
 
         Args:
             to_logger: `Logger` used for logging the redirected messages. If not provided, a new logger is created
             actor_name: Optional component of default logger name.
+            from_start: If `True`, all logs from the start of the actor run will be redirected. If `False`, only newly
+                arrived logs will be redirected. This can be useful for redirecting only a small portion of relevant
+                logs for long-running actors in stand-by.
 
         Returns:
             `StreamedLog` instance for redirected logs.
@@ -274,7 +279,7 @@ def get_streamed_log(self, to_logger: logging.Logger | None = None, actor_name:
             name = '-'.join(part for part in (actor_name, run_id) if part)
             to_logger = create_redirect_logger(f'apify.{name}')
 
-        return StreamedLogSync(log_client=self.log(), to_logger=to_logger)
+        return StreamedLogSync(log_client=self.log(), to_logger=to_logger, from_start=from_start)
 
     def charge(
         self,
@@ -543,14 +548,19 @@ def log(self) -> LogClientAsync:
             **self._sub_resource_init_options(resource_path='log'),
         )
 
-    async def get_streamed_log(self, to_logger: logging.Logger | None = None, actor_name: str = '') -> StreamedLogAsync:
+    async def get_streamed_log(
+        self, to_logger: logging.Logger | None = None, *, actor_name: str = '', from_start: bool = True
+    ) -> StreamedLogAsync:
         """Get `StreamedLog` instance that can be used to redirect logs.
 
          `StreamedLog` can be directly called or used as a context manager.
 
         Args:
             to_logger: `Logger` used for logging the redirected messages. If not provided, a new logger is created
             actor_name: Optional component of default logger name.
+            from_start: If `True`, all logs from the start of the actor run will be redirected. If `False`, only newly
+                arrived logs will be redirected. This can be useful for redirecting only a small portion of relevant
+                logs for long-running actors in stand-by.
 
         Returns:
             `StreamedLog` instance for redirected logs.
@@ -562,7 +572,7 @@ async def get_streamed_log(self, to_logger: logging.Logger | None = None, actor_
             name = '-'.join(part for part in (actor_name, run_id) if part)
             to_logger = create_redirect_logger(f'apify.{name}')
 
-        return StreamedLogAsync(log_client=self.log(), to_logger=to_logger)
+        return StreamedLogAsync(log_client=self.log(), to_logger=to_logger, from_start=from_start)
 
     async def charge(
         self,

tests/unit/test_logging.py

@@ -113,16 +113,20 @@ def propagate_stream_logs() -> None:
     logging.getLogger(f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}').setLevel(logging.DEBUG)
 
 
+@pytest.mark.parametrize(('log_from_start', 'expected_log_count'), [(True, 8), (False, 5)])
 @respx.mock
 async def test_redirected_logs_async(
+    *,
     caplog: LogCaptureFixture,
     mock_api_async: None,  # noqa: ARG001, fixture
     propagate_stream_logs: None,  # noqa: ARG001, fixture
+    log_from_start: bool,
+    expected_log_count: int,
 ) -> None:
     """Test that redirected logs are formatted correctly."""
 
     run_client = ApifyClientAsync(token='mocked_token', api_url=_MOCKED_API_URL).run(run_id=_MOCKED_RUN_ID)
-    streamed_log = await run_client.get_streamed_log(actor_name=_MOCKED_ACTOR_NAME)
+    streamed_log = await run_client.get_streamed_log(actor_name=_MOCKED_ACTOR_NAME, from_start=log_from_start)
 
     # Set `propagate=True` during the tests, so that caplog can see the logs..
     logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
@@ -132,22 +136,26 @@ async def test_redirected_logs_async(
             # Do stuff while the log from the other actor is being redirected to the logs.
             await asyncio.sleep(1)
 
-    assert len(caplog.records) == 8
-    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
+    assert len(caplog.records) == expected_log_count
+    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS[-expected_log_count:], caplog.records):
         assert expected_message_and_level[0] == record.message
         assert expected_message_and_level[1] == record.levelno
 
 
+@pytest.mark.parametrize(('log_from_start', 'expected_log_count'), [(True, 8), (False, 5)])
 @respx.mock
 def test_redirected_logs_sync(
+    *,
     caplog: LogCaptureFixture,
     mock_api_sync: None,  # noqa: ARG001, fixture
     propagate_stream_logs: None,  # noqa: ARG001, fixture
+    log_from_start: bool,
+    expected_log_count: int,
 ) -> None:
     """Test that redirected logs are formatted correctly."""
 
     run_client = ApifyClient(token='mocked_token', api_url=_MOCKED_API_URL).run(run_id=_MOCKED_RUN_ID)
-    streamed_log = run_client.get_streamed_log(actor_name=_MOCKED_ACTOR_NAME)
+    streamed_log = run_client.get_streamed_log(actor_name=_MOCKED_ACTOR_NAME, from_start=log_from_start)
 
     # Set `propagate=True` during the tests, so that caplog can see the logs..
     logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
@@ -156,8 +164,8 @@ def test_redirected_logs_sync(
         # Do stuff while the log from the other actor is being redirected to the logs.
         time.sleep(1)
 
-    assert len(caplog.records) == 8
-    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
+    assert len(caplog.records) == expected_log_count
+    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS[-expected_log_count:], caplog.records):
         assert expected_message_and_level[0] == record.message
         assert expected_message_and_level[1] == record.levelno