Polish spliting of messages and setting the log level

Update test to be able to use caplog

Author: Pijukatel

src/apify_client/_logging.py

@@ -125,33 +125,18 @@ def format(self, record: logging.LogRecord) -> str:
 
 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)
 
-    if respect_original_log_level:
-        to_logger.addFilter(
-            _RedirectLogLevelFilter(
-                default_level=default_level,
-            )
-        )
-    else:
-        to_logger.addFilter(_FixedLevelFilter(default_level=default_level))
-
     to_logger.propagate = False
     handler = logging.StreamHandler()
     handler.setFormatter(RedirectLogFormatter())
@@ -160,40 +145,6 @@ def create_redirect_logger(
     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
-
 class RedirectLogFormatter:
     """Formater applied to default redirect logger."""
 

src/apify_client/clients/resource_clients/log.py

@@ -2,9 +2,10 @@
 
 import asyncio
 import logging
+import re
 from asyncio import Task
 from contextlib import asynccontextmanager, contextmanager
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 from apify_shared.utils import ignore_docs
 
@@ -195,7 +196,7 @@ def __init__(self, log_client: LogClientAsync, to_logger: logging.Logger) -> Non
         self._streaming_task: Task | None = None
 
     def __call__(self) -> Task:
-        """Start the streaming task. The called has to handle any cleanup."""
+        """Start the streaming task. The caller has to handle any cleanup."""
         return asyncio.create_task(self._stream_log(self._to_logger))
 
     async def __aenter__(self) -> Self:
@@ -214,21 +215,33 @@ async def __aexit__(
             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.
-
-                # 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()
+                # Example split marker: \n2025-05-12T15:35:59.429Z
+                date_time_marker_pattern = r"(\n\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z)"
+                splits = re.split(date_time_marker_pattern, data.decode('utf-8'))
+                messages=splits[:1]
+
+                for split_marker, message_without_split_marker in zip(splits[1:-1:2],splits[2::2]):
+                    messages.append(split_marker+message_without_split_marker)
+
+                for message in messages:
+                    to_logger.log(level=self._guess_log_level_from_message(message), msg=message.strip())
+        log_stream.close()
+
+    @staticmethod
+    def _guess_log_level_from_message(message: str) -> int:
+        """Guess the log level from the message."""
+        # Using only levels explicitly mentioned in the 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 logging.INFO

tests/unit/test_logging.py

@@ -15,8 +15,24 @@
 @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_logs_logs = (
+        b"2025-05-13T07:24:12.588Z ACTOR: Pulling Docker image of build.\n"
+        b"2025-05-13T07:24:12.686Z ACTOR: Creating Docker container.\n"
+        b"2025-05-13T07:24:12.745Z ACTOR: Starting Docker container.", # Several logs merged into one message
+        b"2025-05-13T07:24:14.132Z [apify] INFO multiline \n log",
+        b"2025-05-13T07:25:14.132Z [apify] WARNING some warning",
+        b"2025-05-13T07:26:14.132Z [apify] DEBUG c")
     mocked_actor_name = "mocked_actor"
+    mocked_run_id = "mocked_run_id"
+
+    expected_logs_and_levels = [
+        ("2025-05-13T07:24:12.588Z ACTOR: Pulling Docker image of build.", logging.INFO),
+        ("2025-05-13T07:24:12.686Z ACTOR: Creating Docker container.", logging.INFO),
+        ("2025-05-13T07:24:12.745Z ACTOR: Starting Docker container.", logging.INFO),
+        ("2025-05-13T07:24:14.132Z [apify] INFO multiline \n log", logging.INFO),
+        ("2025-05-13T07:25:14.132Z [apify] WARNING some warning", logging.WARNING),
+        ("2025-05-13T07:26:14.132Z [apify] DEBUG c", logging.DEBUG),
+    ]
 
     class AsyncByteStream:
         async def __aiter__(self) -> AsyncIterator[bytes]:
@@ -27,27 +43,25 @@ async def __aiter__(self) -> AsyncIterator[bytes]:
         async def aclose(self) -> None:
             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(
+    respx.get(url=f'https://example.com/v2/actor-runs/{mocked_run_id}').mock(
+        return_value=httpx.Response(content=json.dumps({"data":{'id': mocked_run_id}}),status_code=200))
+    respx.get(url=f'https://example.com/v2/actor-runs/{mocked_run_id}/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')]
+
+    run_client = ApifyClientAsync(token="mocked_token", api_url='https://example.com').run(run_id=mocked_run_id)
     streamed_log = await run_client.get_streamed_log(actor_name=mocked_actor_name)
 
-    with caplog.at_level(logging.DEBUG):
+    # Set `propagate=True` during the tests, so that caplog can see the logs..
+    logger_name = f"apify.{mocked_actor_name}-{mocked_run_id}"
+    logging.getLogger(logger_name).propagate = True
+
+    with caplog.at_level(logging.DEBUG, logger=logger_name):
         async with streamed_log:
+            # Do stuff while the log from the other actor is being redirected to the logs.
             await asyncio.sleep(1)
-            # do some stuff
-            pass
 
     records = caplog.records
-    assert len(records) == 2
-
-
-
-"""
-
-{'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'}
-"""
+    assert len(records) == 6
+    for expected_log_and_level, record in zip(expected_logs_and_levels, records):
+        assert expected_log_and_level[0] == record.message
+        assert expected_log_and_level[1] == record.levelno