WIP

TODO: Message inspection to split bulk messages
TODO: Log level guessing outside of filter

Author: Pijukatel

src/apify_client/_logging.py

@@ -7,6 +7,8 @@
 from contextvars import ContextVar
 from typing import TYPE_CHECKING, Any, Callable, NamedTuple, cast
 
+from colorama import Style, Fore
+
 # 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
@@ -121,7 +123,6 @@ def format(self, record: logging.LogRecord) -> str:
             log_string = f'{log_string} ({json.dumps(extra)})'
         return log_string
 
-
 def create_redirect_logger(
     name: str,
     *,
@@ -141,7 +142,7 @@ def create_redirect_logger(
         The created logger.
     """
     to_logger = logging.getLogger(name)
-    to_logger.propagate = True
+
     if respect_original_log_level:
         to_logger.addFilter(
             _RedirectLogLevelFilter(
@@ -151,6 +152,11 @@ def create_redirect_logger(
     else:
         to_logger.addFilter(_FixedLevelFilter(default_level=default_level))
 
+    to_logger.propagate = False
+    handler = logging.StreamHandler()
+    handler.setFormatter(RedirectLogFormatter())
+    to_logger.addHandler(handler)
+    to_logger.setLevel(logging.INFO)
     return to_logger
 
 
@@ -187,3 +193,17 @@ def filter(self, record: logging.LogRecord) -> bool:
         record.levelno = self._guess_log_level_from_message(record.msg)
         record.levelname = logging.getLevelName(record.levelno)
         return True
+
+class RedirectLogFormatter:
+    """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/log.py

@@ -224,6 +224,11 @@ async def _stream_log(self, to_logger: logging.Logger) -> None:
             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)
+
+                # Split by lines for each line that does start with standard format, try to guess the log level
+                # example split marker: \n2025-05-12T15:35:59.429Z
+
+                to_logger.log(level=log_level, msg=data.decode('utf-8'))
+                #logging.getLogger("apify_client").info(data)
         # Cleanup in the end
         #log_stream.close()

src/apify_client/clients/resource_clients/run.py

@@ -533,7 +533,8 @@ async def get_streamed_log(self, to_logger: logging.Logger | None = None, actor_
         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}"}')
+            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)
 

tests/unit/test_logging.py

@@ -1,3 +1,6 @@
+import asyncio
+import json
+import logging
 import time
 from collections.abc import AsyncIterator
 
@@ -9,31 +12,42 @@
 from apify_client.clients import RunClientAsync
 
 
-
-
 @respx.mock
 async def test_redirected_logs(caplog) -> None:
     """Test that redirected logs are formatted correctly."""
+    mocked_actor_logs_logs = (b"INFO a", b"WARNING b", b"DEBUG c")
+    mocked_actor_name = "mocked_actor"
 
     class AsyncByteStream:
         async def __aiter__(self) -> AsyncIterator[bytes]:
-            for i in range(2):
-                yield b"Some text"
-                time.sleep(1)
-            print("b")
+            for i in mocked_actor_logs_logs:
+                yield i
+                await asyncio.sleep(0.1)
 
         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(content=json.dumps({"data":{'actId': 'SbjD4JEucMevUdQAH'}}),status_code=200))
+    respx.get(url='https://example.com/v2/actor-runs/run_is_mocked/log?stream=1').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
+    streamed_log = await run_client.get_streamed_log(actor_name=mocked_actor_name)
+
+    with caplog.at_level(logging.DEBUG):
+        async with streamed_log:
+            await asyncio.sleep(1)
+            # do some stuff
+            pass
+
+    records = caplog.records
+    assert len(records) == 2
+
+
+
+"""
 
-    records = caplog.get_records()
+{'actId': 'SbjD4JEucMevUdQAH', 'buildId': 'Jv7iIjo1JV0gEXQEm', 'buildNumber': '0.0.5', 'containerUrl': 'https://tlo2axp6qbc7.runs.apify.net', 'defaultDatasetId': 'DZq6uDwZ4gSXev8h2', 'defaultKeyValueStoreId': '7UswAGyvNKFGlddHS', 'defaultRequestQueueId': 'Gk4ye89GRCoqFNdsM', 'finishedAt': None, 'generalAccess': 'FOLLOW_USER_SETTING', 'id': 'u6Q52apBHWO09NjDP', 'meta': {'origin': 'API', 'userAgent': 'ApifyClient/1.9.0 (linux; Python/3.10.12); isAtHome/False'}, 'options': {'build': 'latest', 'diskMbytes': 2048, 'memoryMbytes': 1024, 'timeoutSecs': 3600}, 'startedAt': '2025-05-12T13:54:23.028Z', 'stats': {'computeUnits': 0, 'inputBodyLen': 15, 'migrationCount': 0, 'rebootCount': 0, 'restartCount': 0, 'resurrectCount': 0}, 'status': 'READY', 'usage': {'ACTOR_COMPUTE_UNITS': 0, 'DATASET_READS': 0, 'DATASET_WRITES': 0, 'DATA_TRANSFER_EXTERNAL_GBYTES': 0, 'DATA_TRANSFER_INTERNAL_GBYTES': 0, 'KEY_VALUE_STORE_LISTS': 0, 'KEY_VALUE_STORE_READS': 0, 'KEY_VALUE_STORE_WRITES': 1, 'PROXY_RESIDENTIAL_TRANSFER_GBYTES': 0, 'PROXY_SERPS': 0, 'REQUEST_QUEUE_READS': 0, 'REQUEST_QUEUE_WRITES': 0}, 'usageTotalUsd': 5e-05, 'usageUsd': {'ACTOR_COMPUTE_UNITS': 0, 'DATASET_READS': 0, 'DATASET_WRITES': 0, 'DATA_TRANSFER_EXTERNAL_GBYTES': 0, 'DATA_TRANSFER_INTERNAL_GBYTES': 0, 'KEY_VALUE_STORE_LISTS': 0, 'KEY_VALUE_STORE_READS': 0, 'KEY_VALUE_STORE_WRITES': 5e-05, 'PROXY_RESIDENTIAL_TRANSFER_GBYTES': 0, 'PROXY_SERPS': 0, 'REQUEST_QUEUE_READS': 0, 'REQUEST_QUEUE_WRITES': 0}, 'userId': 'LjAzEG1CadliECnrn'}
+"""