Add sync version of the logging.

Refactor tests and code to reuse code.

Author: Pijukatel

src/apify_client/clients/resource_clients/actor.py

@@ -332,8 +332,19 @@ def call(
             timeout_secs=timeout_secs,
             webhooks=webhooks,
         )
+        if not logger:
+            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':
+            actor_data = self.get()
+            actor_name = actor_data.get('name', '') if actor_data else ''
+            log_context = run_client.get_streamed_log(actor_name=actor_name)
+        else:
+            log_context = run_client.get_streamed_log(to_logger=logger)
 
-        return self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
+        with log_context:
+            return self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
     def build(
         self,

src/apify_client/clients/resource_clients/log.py

@@ -3,8 +3,10 @@
 import asyncio
 import logging
 import re
+import threading
 from asyncio import Task
 from contextlib import asynccontextmanager, contextmanager
+from threading import Thread
 from typing import TYPE_CHECKING, Any, cast
 
 from apify_shared.utils import ignore_docs
@@ -201,67 +203,33 @@ async def stream(self, *, raw: bool = False) -> AsyncIterator[httpx.Response | N
                 await response.aclose()
 
 
-class StreamedLogSync:
-    """Utility class for streaming logs from another actor."""
-
-
-class StreamedLogAsync:
+class StreamedLog:
     """Utility class for streaming logs from another actor."""
 
     # Test related flag to enable propagation of logs to the `caplog` fixture during tests.
     _force_propagate = False
 
-    def __init__(self, log_client: LogClientAsync, to_logger: logging.Logger) -> None:
-        self._log_client = log_client
+    def __init__(self, to_logger: logging.Logger) -> None:
         self._to_logger = to_logger
-        self._streaming_task: Task | None = None
         if self._force_propagate:
             to_logger.propagate = True
         self._stream_buffer = list[str]()
         # Redirected logs are forwarded to logger as soon as there are at least two split markers present in the buffer.
         # For example, 2025-05-12T15:35:59.429Z
         self._split_marker = re.compile(r'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z)')
 
-    def __call__(self) -> Task:
-        """Start the streaming task. The caller has to handle any cleanup."""
-        return asyncio.create_task(self._stream_log())
-
-    async def __aenter__(self) -> Self:
-        """Start the streaming task within the context. Exiting the context will cancel the streaming task."""
-        if self._streaming_task:
-            raise RuntimeError('Streaming task already active')
-        self._streaming_task = self()
-
-        return self
-
-    async def __aexit__(
-        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
-    ) -> None:
-        """Cancel the streaming task."""
-        if not self._streaming_task:
-            raise RuntimeError('Streaming task is not active')
-
-        self._streaming_task.cancel()
-        self._streaming_task = None
-
-    async def _stream_log(self) -> None:
-        async with self._log_client.stream(raw=True) as log_stream:
-            if not log_stream:
-                return
-            async for data in log_stream.aiter_bytes():
-                new_chunk = data.decode('utf-8')
-                self._stream_buffer.append(new_chunk)
-                if re.findall(self._split_marker, new_chunk):
-                    # If complete split marker was found in new chunk, then process the buffer.
-                    self._log_buffer_content(include_last_part=False)
-
-            # If the stream is finished, then the last part will be also processed.
-            self._log_buffer_content(include_last_part=True)
+    def _process_new_data(self, data: bytes) -> None:
+        new_chunk = data.decode('utf-8')
+        self._stream_buffer.append(new_chunk)
+        if re.findall(self._split_marker, new_chunk):
+            # If complete split marker was found in new chunk, then process the buffer.
+            self._log_buffer_content(include_last_part=False)
 
     def _log_buffer_content(self, *, include_last_part: bool = False) -> None:
-        """Merge the whole buffer and plit it into parts based on the marker.
+        """Merge the whole buffer and split it into parts based on the marker.
 
-        The last part could be incomplete, and so it can be left unprocessed and in the buffer.
+        Log the messages created from the split parts and remove them from buffer.
+        The last part could be incomplete, and so it can be left unprocessed and in the buffer until later.
         """
         all_parts = re.split(self._split_marker, ''.join(self._stream_buffer))
         # First split is empty string
@@ -290,3 +258,98 @@ def _guess_log_level_from_message(message: str) -> int:
                 return cast('int', logging.getLevelName(level))
         # Unknown log level. Fall back to the default.
         return logging.INFO
+
+
+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)
+        self._log_client = log_client
+        self._streaming_thread: Thread | None = None
+        self._stop_logging = False
+
+    def __call__(self) -> Thread:
+        """Start the streaming thread. The caller has to handle any cleanup."""
+        if self._streaming_thread:
+            raise RuntimeError('Streaming thread already active')
+        self._stop_logging = False
+        self._streaming_thread = threading.Thread(target=self._stream_log)
+        self._streaming_thread.start()
+        return self._streaming_thread
+
+    def __enter__(self) -> Self:
+        """Start the streaming thread within the context. Exiting the context will finish the streaming thread."""
+        self()
+        return self
+
+    def __exit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> None:
+        """Stop the streaming thread."""
+        if not self._streaming_thread:
+            raise RuntimeError('Streaming thread is not active')
+
+        # Signal the thread to stop logging and wait for it to finish.
+        self._stop_logging = True
+        self._streaming_thread.join()
+        self._streaming_thread = None
+        self._stop_logging = False
+
+    def _stream_log(self) -> None:
+        with self._log_client.stream(raw=True) as log_stream:
+            if not log_stream:
+                return
+            for data in log_stream.iter_bytes():
+                new_chunk = data.decode('utf-8')
+                self._stream_buffer.append(new_chunk)
+                if re.findall(self._split_marker, new_chunk):
+                    # If complete split marker was found in new chunk, then process the buffer.
+                    self._log_buffer_content(include_last_part=False)
+                if self._stop_logging:
+                    break
+
+            # If the stream is finished, then the last part will be also processed.
+            self._log_buffer_content(include_last_part=True)
+        return
+
+
+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)
+        self._log_client = log_client
+        self._streaming_task: Task | None = None
+
+    def __call__(self) -> Task:
+        """Start the streaming task. The caller has to handle any cleanup."""
+        if self._streaming_task:
+            raise RuntimeError('Streaming task already active')
+        self._streaming_task = asyncio.create_task(self._stream_log())
+        return self._streaming_task
+
+    async def __aenter__(self) -> Self:
+        """Start the streaming task within the context. Exiting the context will cancel the streaming task."""
+        self()
+        return self
+
+    async def __aexit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> None:
+        """Cancel the streaming task."""
+        if not self._streaming_task:
+            raise RuntimeError('Streaming task is not active')
+
+        self._streaming_task.cancel()
+        self._streaming_task = None
+
+    async def _stream_log(self) -> None:
+        async with self._log_client.stream(raw=True) as log_stream:
+            if not log_stream:
+                return
+            async for data in log_stream.aiter_bytes():
+                self._process_new_data(data)
+
+            # If the stream is finished, then the last part will be also processed.
+            self._log_buffer_content(include_last_part=True)

src/apify_client/clients/resource_clients/run.py

@@ -13,7 +13,12 @@
 from apify_client.clients.base import ActorJobBaseClient, ActorJobBaseClientAsync
 from apify_client.clients.resource_clients.dataset import DatasetClient, DatasetClientAsync
 from apify_client.clients.resource_clients.key_value_store import KeyValueStoreClient, KeyValueStoreClientAsync
-from apify_client.clients.resource_clients.log import LogClient, LogClientAsync, StreamedLogAsync
+from apify_client.clients.resource_clients.log import (
+    LogClient,
+    LogClientAsync,
+    StreamedLogAsync,
+    StreamedLogSync,
+)
 from apify_client.clients.resource_clients.request_queue import RequestQueueClient, RequestQueueClientAsync
 
 if TYPE_CHECKING:
@@ -250,6 +255,27 @@ 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:
+        """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.
+
+        Returns:
+            `StreamedLog` instance for redirected logs.
+        """
+        run_data = self.get()
+        run_id = run_data.get('id', '') 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 StreamedLogSync(log_client=self.log(), to_logger=to_logger)
+
     def charge(
         self,
         event_name: str,
@@ -518,16 +544,16 @@ def log(self) -> LogClientAsync:
         )
 
     async def get_streamed_log(self, to_logger: logging.Logger | None = None, actor_name: str = '') -> StreamedLogAsync:
-        """Get `StreamedLogAsync` instance that can be used to redirect logs.
+        """Get `StreamedLog` instance that can be used to redirect logs.
 
-         `StreamedLogAsync` can be directly called or used as a context manager.
+         `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.
 
         Returns:
-            `StreamedLogAsync` instance for redirected logs.
+            `StreamedLog` instance for redirected logs.
         """
         run_data = await self.get()
         run_id = run_data.get('id', '') if run_data else ''
@@ -536,7 +562,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(self.log(), to_logger)
+        return StreamedLogAsync(log_client=self.log(), to_logger=to_logger)
 
     async def charge(
         self,