TODO: Figure out hwo to mock response with steram

Author: Pijukatel

src/apify_client/_logging.py

@@ -5,7 +5,7 @@
 import json
 import logging
 from contextvars import ContextVar
-from typing import TYPE_CHECKING, Any, Callable, NamedTuple
+from typing import TYPE_CHECKING, Any, Callable, NamedTuple, cast
 
 # Conditional import only executed when type checking, otherwise we'd get circular dependency issues
 if TYPE_CHECKING:
@@ -120,3 +120,70 @@ 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,
+    *,
+    default_level: int = logging.INFO,
+    respect_original_log_level: bool = True,
+) -> 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.
+        default_level: All logs will be emitted on this level, regardless of the original level.
+        respect_original_log_level: Try to use the original log level of the other actor logger and use `default_level`
+         only if the original level is not determined.
+
+    Returns:
+        The created logger.
+    """
+    to_logger = logging.getLogger(name)
+    to_logger.propagate = True
+    if respect_original_log_level:
+        to_logger.addFilter(
+            _RedirectLogLevelFilter(
+                default_level=default_level,
+            )
+        )
+    else:
+        to_logger.addFilter(_FixedLevelFilter(default_level=default_level))
+
+    return to_logger
+
+
+class _FixedLevelFilter(logging.Filter):
+    def __init__(self, name: str = '', default_level: int = logging.INFO) -> None:
+        super().__init__(name)
+        self._default_level = default_level
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Try to extract log level from the message content and set it accordingly."""
+        record.levelno = self._default_level
+        record.levelname = logging.getLevelName(record.levelno)
+        return True
+
+
+class _RedirectLogLevelFilter(logging.Filter):
+    def __init__(self, name: str = '', default_level: int = logging.INFO) -> None:
+        super().__init__(name)
+        self._default_level = default_level
+
+    def _guess_log_level_from_message(self, message: str) -> int:
+        """Guess the log level from the message."""
+        # Using only levels explicitly mentioned in logging module
+        known_levels = ('CRITICAL', 'FATAL', 'ERROR', 'WARN', 'WARNING', 'INFO', 'DEBUG', 'NOTSET')
+        for level in known_levels:
+            if level in message:
+                # `getLevelName` returns an `int` when string is passed as input.
+                return cast('int', logging.getLevelName(level))
+        # Unknown log level. Fall back to the default.
+        return self._default_level
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Try to extract log level from the message content and set it accordingly."""
+        record.levelno = self._guess_log_level_from_message(record.msg)
+        record.levelname = logging.getLevelName(record.levelno)
+        return True

src/apify_client/clients/resource_clients/log.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import asyncio
+import logging
+from asyncio import Task
 from contextlib import asynccontextmanager, contextmanager
 from typing import TYPE_CHECKING, Any
 
@@ -11,8 +14,10 @@
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Iterator
+    from types import TracebackType
 
     import httpx
+    from mypy.types import Self
 
 
 class LogClient(ResourceClient):
@@ -175,3 +180,50 @@ async def stream(self) -> AsyncIterator[httpx.Response | None]:
         finally:
             if response:
                 await response.aclose()
+
+
+class StreamedLogSync:
+    """Utility class for streaming logs from another actor."""
+
+
+class StreamedLogAsync:
+    """Utility class for streaming logs from another actor."""
+
+    def __init__(self, log_client: LogClientAsync, to_logger: logging.Logger) -> None:
+        self._log_client = log_client
+        self._to_logger = to_logger
+        self._streaming_task: Task | None = None
+
+    def __call__(self) -> Task:
+        """Start the streaming task. The called has to handle any cleanup."""
+        return asyncio.create_task(self._stream_log(self._to_logger))
+
+    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, to_logger: logging.Logger) -> None:
+        async with self._log_client.stream() as log_stream:
+            if not log_stream:
+                return
+            async for data in log_stream.aiter_bytes():
+                log_level = logging.INFO  # The Original log level is not known unless the message is inspected.
+                # Adjust the log level in custom logger filter if needed.
+                to_logger.log(level=log_level, msg=data)
+        # Cleanup in the end
+        #log_stream.close()

src/apify_client/clients/resource_clients/run.py

@@ -8,14 +8,16 @@
 
 from apify_shared.utils import filter_out_none_values_recursively, ignore_docs, parse_date_fields
 
+from apify_client._logging import create_redirect_logger
 from apify_client._utils import encode_key_value_store_record_value, pluck_data, to_safe_id
 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
+from apify_client.clients.resource_clients.log import LogClient, LogClientAsync, StreamedLogAsync
 from apify_client.clients.resource_clients.request_queue import RequestQueueClient, RequestQueueClientAsync
 
 if TYPE_CHECKING:
+    import logging
     from decimal import Decimal
 
     from apify_shared.consts import RunGeneralAccess
@@ -515,6 +517,26 @@ 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:
+        """Get `StreamedLogAsync` instance that can be used to redirect logs.
+
+         `StreamedLogAsync` 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.
+        """
+        run_data = await self.get()
+        run_id = run_data.get('id', '') if run_data else ''
+
+        if not to_logger:
+            to_logger = create_redirect_logger(f'apify.{f"{actor_name}-{run_id}"}')
+
+        return StreamedLogAsync(self.log(), to_logger)
+
     async def charge(
         self,
         event_name: str,

tests/unit/test_logging.py

@@ -0,0 +1,39 @@
+import time
+from collections.abc import AsyncIterator
+
+import httpx
+import respx
+
+
+from apify_client import ApifyClientAsync
+from apify_client.clients import RunClientAsync
+
+
+
+
+@respx.mock
+async def test_redirected_logs(caplog) -> None:
+    """Test that redirected logs are formatted correctly."""
+
+    class AsyncByteStream:
+        async def __aiter__(self) -> AsyncIterator[bytes]:
+            for i in range(2):
+                yield b"Some text"
+                time.sleep(1)
+            print("b")
+
+        async def aclose(self) -> None:
+            print("a")
+            pass
+
+    run_client = ApifyClientAsync(token="mocked_token", api_url='https://example.com').run(run_id="run_is_mocked")
+    respx.get(url='https://example.com/v2/actor-runs/run_is_mocked').mock(
+        return_value=httpx.Response(stream=AsyncByteStream(), status_code=200))
+        # {'http_version': b'HTTP/1.1', 'network_stream': <httpcore._backends.anyio.AnyIOStream object at 0x7fc82543db70>, 'reason_phrase': b'OK'}
+        # [(b'Date', b'Mon, 12 May 2025 13:24:41 GMT'), (b'Content-Type', b'application/json; charset=utf-8'), (b'Transfer-Encoding', b'chunked'), (b'Connection', b'keep-alive'), (b'Cache-Control', b'no-cache, no-store, must-revalidate'), (b'Pragma', b'no-cache'), (b'Expires', b'0'), (b'Access-Control-Allow-Origin', b'*'), (b'Access-Control-Allow-Headers', b'User-Agent, Content-Type, Authorization, X-Apify-Request-Origin, openai-conversation-id, openai-ephemeral-user-id'), (b'Access-Control-Allow-Methods', b'GET, POST'), (b'Access-Control-Expose-Headers', b'X-Apify-Pagination-Total, X-Apify-Pagination-Offset, X-Apify-Pagination-Desc, X-Apify-Pagination-Count, X-Apify-Pagination-Limit'), (b'Referrer-Policy', b'no-referrer'), (b'X-Robots-Tag', b'none'), (b'X-RateLimit-Limit', b'200'), (b'Location', b'https://api.apify.com/v2/actor-runs/ywNUnFFbOksQLa4mH'), (b'Vary', b'Accept-Encoding'), (b'Content-Encoding', b'gzip')]
+    streamed_log = await run_client.get_streamed_log(actor_name="mocked_actor")
+    async with streamed_log:
+        # do some stuff
+        pass
+
+    records = caplog.get_records()