Self review.

Have explicit start and stop instead of __call__

Author: Pijukatel

src/apify_client/_logging.py

@@ -158,10 +158,11 @@ class RedirectLogFormatter(logging.Formatter):
     def format(self, record: logging.LogRecord) -> str:
         """Format the log by prepending logger name to the original message.
 
-        TODO: Make advanced coloring later.
-        Ideally it should respect the color of the original log, but that information is not available in the API.
-        Inspecting logs and coloring their parts during runtime could be quite heavy. Keep it simple for now.
-        """
-        logger_name_string = f'{Fore.CYAN}[{record.name}]{Style.RESET_ALL} '
+        Args:
+            record: Log record to be formated.
 
-        return f'{logger_name_string}-> {record.msg}'
+        Returns:
+            Formated log message.
+        """
+        formated_logger_name = f'{Fore.CYAN}[{record.name}]{Style.RESET_ALL} '
+        return f'{formated_logger_name}-> {record.msg}'

src/apify_client/clients/resource_clients/log.py

@@ -222,13 +222,11 @@ def __init__(self, to_logger: logging.Logger, *, from_start: bool = True) -> Non
         """Initialize `StreamedLog`.
 
         Args:
-            to_logger: The logger to which the logs will be redirected
+            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:
@@ -248,7 +246,7 @@ def _log_buffer_content(self, *, include_last_part: bool = False) -> None:
         """Merge the whole buffer and split it into parts based on the marker.
 
         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.
+        The last part could be incomplete, and so it can be left unprocessed in the buffer until later.
         """
         all_parts = re.split(self._split_marker, ''.join(self._stream_buffer))[1:]  # First split is empty string
         if include_last_part:
@@ -292,32 +290,34 @@ def __init__(self, log_client: LogClient, *, to_logger: logging.Logger, from_sta
         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."""
+    def start(self) -> Thread:
+        """Start the streaming thread. The caller has to handle any cleanup by manually calling the `stop` method."""
         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 stop(self) -> None:
+        """Signal the streaming thread to stop logging and wait for it to finish."""
+        if not self._streaming_thread:
+            raise RuntimeError('Streaming thread is not active')
+        self._stop_logging = True
+        self._streaming_thread.join()
+        self._streaming_thread = None
+        self._stop_logging = False
+
     def __enter__(self) -> Self:
         """Start the streaming thread within the context. Exiting the context will finish the streaming thread."""
-        self()
+        self.start()
         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
+        self.stop()
 
     def _stream_log(self) -> None:
         with self._log_client.stream(raw=True) as log_stream:
@@ -341,27 +341,31 @@ def __init__(self, log_client: LogClientAsync, *, to_logger: logging.Logger, fro
         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."""
+    def start(self) -> Task:
+        """Start the streaming task. The caller has to handle any cleanup by manually calling the `stop` method."""
         if self._streaming_task:
             raise RuntimeError('Streaming task already active')
         self._streaming_task = asyncio.create_task(self._stream_log())
         return self._streaming_task
 
+    def stop(self) -> None:
+        """Stop 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 __aenter__(self) -> Self:
         """Start the streaming task within the context. Exiting the context will cancel the streaming task."""
-        self()
+        self.start()
         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
+        self.stop()
 
     async def _stream_log(self) -> None:
         async with self._log_client.stream(raw=True) as log_stream: