feat: Add redirected actor logs

pyproject.toml

@@ -25,6 +25,7 @@ classifiers = [
 keywords = ["apify", "api", "client", "automation", "crawling", "scraping"]
 dependencies = [
     "apify-shared>=1.4.1",
+    "colorama~=0.4.0",
     "httpx>=0.25",
     "more_itertools>=10.0.0",
 ]
@@ -52,6 +53,7 @@ dev = [
     "respx~=0.22.0",
     "ruff~=0.11.0",
     "setuptools",  # setuptools are used by pytest but not explicitly required
+    "types-colorama~=0.4.15.20240106",
 ]
 
 [tool.hatch.build.targets.wheel]

src/apify_client/_logging.py

@@ -7,6 +7,8 @@
 from contextvars import ContextVar
 from typing import TYPE_CHECKING, Any, Callable, NamedTuple
 
+from colorama import Fore, Style
+
 # Conditional import only executed when type checking, otherwise we'd get circular dependency issues
 if TYPE_CHECKING:
     from apify_client.clients.base.base_client import _BaseBaseClient
@@ -120,3 +122,46 @@ def format(self, record: logging.LogRecord) -> str:
         if extra:
             log_string = f'{log_string} ({json.dumps(extra)})'
         return log_string
+
+
+def create_redirect_logger(
+    name: str,
+) -> logging.Logger:
+    """Create a logger for redirecting logs from another Actor.
+
+    Args:
+        name: The name of the logger. It can be used to inherit from other loggers. Example: `apify.xyz` will use logger
+         named `xyz` and make it a children of `apify` logger.
+
+    Returns:
+        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)
+    to_logger.setLevel(logging.DEBUG)
+    return to_logger
+
+
+class RedirectLogFormatter(logging.Formatter):
+    """Formater applied to default redirect logger."""
+
+    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} '
+
+        return f'{logger_name_string}-> {record.msg}'

src/apify_client/clients/resource_clients/actor.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal
 
 from apify_shared.utils import (
     filter_out_none_values_recursively,
@@ -27,6 +27,7 @@
 
 if TYPE_CHECKING:
     from decimal import Decimal
+    from logging import Logger
 
     from apify_shared.consts import ActorJobStatus, MetaOrigin
 
@@ -289,6 +290,7 @@ def call(
         timeout_secs: int | None = None,
         webhooks: list[dict] | None = None,
         wait_secs: int | None = None,
+        logger: Logger | None | Literal['default'] = 'default',
     ) -> dict | None:
         """Start the Actor and wait for it to finish before returning the Run object.
 
@@ -313,6 +315,9 @@ def call(
                 a webhook set up for the Actor, you do not have to add it again here.
             wait_secs: The maximum number of seconds the server waits for the run to finish. If not provided,
                 waits indefinitely.
+            logger: Loger used to redirect logs from the Actor run. By default, it is set to "default" which means that
+                the default logger will be created and used. Setting `None` will disable any log propagation. Passing
+                custom logger will redirect logs to the provided logger.
 
         Returns:
             The run object.
@@ -327,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)
 
-        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)
+
+        with log_context:
+            return self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
     def build(
         self,
@@ -681,6 +697,7 @@ async def call(
         timeout_secs: int | None = None,
         webhooks: list[dict] | None = None,
         wait_secs: int | None = None,
+        logger: Logger | None | Literal['default'] = 'default',
     ) -> dict | None:
         """Start the Actor and wait for it to finish before returning the Run object.
 
@@ -705,6 +722,9 @@ async def call(
                 a webhook set up for the Actor, you do not have to add it again here.
             wait_secs: The maximum number of seconds the server waits for the run to finish. If not provided,
                 waits indefinitely.
+            logger: Loger used to redirect logs from the Actor run. By default, it is set to "default" which means that
+                the default logger will be created and used. Setting `None` will disable any log propagation. Passing
+                custom logger will redirect logs to the provided logger.
 
         Returns:
             The run object.
@@ -720,7 +740,19 @@ async def call(
             webhooks=webhooks,
         )
 
-        return await self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
+        if not logger:
+            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':
+            actor_data = await self.get()
+            actor_name = actor_data.get('name', '') if actor_data else ''
+            log_context = await run_client.get_streamed_log(actor_name=actor_name)
+        else:
+            log_context = await run_client.get_streamed_log(to_logger=logger)
+
+        async with log_context:
+            return await self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
     async def build(
         self,