Draft with async implementation and example tests

Author: Pijukatel

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

@@ -5,9 +5,9 @@
 import json
 import logging
 from contextvars import ContextVar
-from typing import TYPE_CHECKING, Any, Callable, NamedTuple, cast
+from typing import TYPE_CHECKING, Any, Callable, NamedTuple
 
-from colorama import Style, Fore
+from colorama import Fore, Style
 
 # Conditional import only executed when type checking, otherwise we'd get circular dependency issues
 if TYPE_CHECKING:
@@ -123,6 +123,7 @@ def format(self, record: logging.LogRecord) -> str:
             log_string = f'{log_string} ({json.dumps(extra)})'
         return log_string
 
+
 def create_redirect_logger(
     name: str,
 ) -> logging.Logger:
@@ -141,11 +142,11 @@ def create_redirect_logger(
     handler = logging.StreamHandler()
     handler.setFormatter(RedirectLogFormatter())
     to_logger.addHandler(handler)
-    to_logger.setLevel(logging.INFO)
+    to_logger.setLevel(logging.DEBUG)
     return to_logger
 
 
-class RedirectLogFormatter:
+class RedirectLogFormatter(logging.Formatter):
     """Formater applied to default redirect logger."""
 
     def format(self, record: logging.LogRecord) -> str:

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
 
@@ -681,6 +682,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 +707,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 +725,18 @@ 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_name = actor_data.get('name', '') if (actor_data := await self.get()) 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,

src/apify_client/clients/resource_clients/log.py

@@ -190,10 +190,15 @@ class StreamedLogSync:
 class StreamedLogAsync:
     """Utility class for streaming logs from another actor."""
 
+    # Test related flag to enable propagation of logs to the `caplog` fixture during tests.
+    _force_propagate = False
+
     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
+        if self._force_propagate:
+            to_logger.propagate = True
 
     def __call__(self) -> Task:
         """Start the streaming task. The caller has to handle any cleanup."""
@@ -223,12 +228,12 @@ async def _stream_log(self, to_logger: logging.Logger) -> None:
                 return
             async for data in log_stream.aiter_bytes():
                 # 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)"
+                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]
+                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 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())

src/apify_client/clients/resource_clients/run.py

@@ -533,7 +533,7 @@ 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:
-            name = "-".join(part for part in (actor_name, run_id) if part)
+            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,67 +1,174 @@
 import asyncio
 import json
 import logging
-import time
 from collections.abc import AsyncIterator
 
 import httpx
+import pytest
 import respx
-
+from _pytest.logging import LogCaptureFixture
+from apify_shared.consts import ActorJobStatus
 
 from apify_client import ApifyClientAsync
-from apify_client.clients import RunClientAsync
+from apify_client._logging import RedirectLogFormatter
+from apify_client.clients.resource_clients.log import StreamedLogAsync
 
+_MOCKED_API_URL = 'https://example.com'
+_MOCKED_RUN_ID = 'mocked_run_id'
+_MOCKED_ACTOR_NAME = 'mocked_actor_name'
+_MOCKED_ACTOR_ID = 'mocked_actor_id'
+_MOCKED_ACTOR_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',
+)
 
-@respx.mock
-async def test_redirected_logs(caplog) -> None:
-    """Test that redirected logs are formatted correctly."""
-    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:
+_EXPECTED_MESSAGES_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),
+)
+
+
+@pytest.fixture
+def mock_api() -> None:
+    class AsyncByteStream(httpx._types.AsyncByteStream):
         async def __aiter__(self) -> AsyncIterator[bytes]:
-            for i in mocked_actor_logs_logs:
+            for i in _MOCKED_ACTOR_LOGS:
                 yield i
                 await asyncio.sleep(0.1)
 
         async def aclose(self) -> None:
             pass
 
-    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))
+    actor_runs_responses = iter(
+        (
+            httpx.Response(
+                content=json.dumps({'data': {'id': _MOCKED_RUN_ID, 'status': ActorJobStatus.RUNNING}}), status_code=200
+            ),
+            httpx.Response(
+                content=json.dumps({'data': {'id': _MOCKED_RUN_ID, 'status': ActorJobStatus.RUNNING}}), status_code=200
+            ),
+            httpx.Response(
+                content=json.dumps({'data': {'id': _MOCKED_RUN_ID, 'status': ActorJobStatus.SUCCEEDED}}),
+                status_code=200,
+            ),
+        )
+    )
+
+    async def actor_runs_side_effect(_: httpx.Request) -> httpx.Response:
+        await asyncio.sleep(0.5)
+        return next(actor_runs_responses)
+
+    respx.get(url=f'{_MOCKED_API_URL}/v2/actor-runs/{_MOCKED_RUN_ID}').mock(side_effect=actor_runs_side_effect)
+
+    respx.get(url=f'{_MOCKED_API_URL}/v2/acts/{_MOCKED_ACTOR_ID}').mock(
+        return_value=httpx.Response(content=json.dumps({'data': {'name': _MOCKED_ACTOR_NAME}}), status_code=200)
+    )
+
+    respx.post(url=f'{_MOCKED_API_URL}/v2/acts/{_MOCKED_ACTOR_ID}/runs').mock(
+        return_value=httpx.Response(content=json.dumps({'data': {'id': _MOCKED_RUN_ID}}), status_code=200)
+    )
 
-    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)
+    respx.get(url=f'{_MOCKED_API_URL}/v2/actor-runs/{_MOCKED_RUN_ID}/log?stream=1').mock(
+        return_value=httpx.Response(stream=AsyncByteStream(), status_code=200)
+    )
+
+
+@pytest.fixture
+def propagate_stream_logs() -> None:
+    StreamedLogAsync._force_propagate = True  # Enable propagation of logs to the caplog fixture
+    logging.getLogger(f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}').setLevel(logging.DEBUG)
+
+
+@respx.mock
+async def test_redirected_logs(
+    caplog: LogCaptureFixture,
+    mock_api: None,  # noqa: ARG001, fixture
+    propagate_stream_logs: None,  # noqa: ARG001, fixture
+) -> None:
+    """Test that redirected logs are formatted correctly."""
+
+    run_client = ApifyClientAsync(token='mocked_token', api_url=_MOCKED_API_URL).run(run_id=_MOCKED_RUN_ID)
+    streamed_log = await run_client.get_streamed_log(actor_name=_MOCKED_ACTOR_NAME)
 
     # 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
+    logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
 
     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)
 
-    records = caplog.records
-    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
+    assert len(caplog.records) == 6
+    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
+        assert expected_message_and_level[0] == record.message
+        assert expected_message_and_level[1] == record.levelno
+
+
+@respx.mock
+async def test_actor_call_redirect_logs_to_default_logger(
+    caplog: LogCaptureFixture,
+    mock_api: None,  # noqa: ARG001, fixture
+    propagate_stream_logs: None,  # noqa: ARG001, fixture
+) -> None:
+    """Test that logs are redirected correctly to the default logger.
+
+    Caplog contains logs before formatting, so formatting is not included in the test expectations."""
+    logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
+    logger = logging.getLogger(logger_name)
+    run_client = ApifyClientAsync(token='mocked_token', api_url=_MOCKED_API_URL).actor(actor_id=_MOCKED_ACTOR_ID)
+
+    with caplog.at_level(logging.DEBUG, logger=logger_name):
+        await run_client.call()
+
+    # Ensure expected handler and formater
+    assert isinstance(logger.handlers[0].formatter, RedirectLogFormatter)
+    assert isinstance(logger.handlers[0], logging.StreamHandler)
+
+    # Ensure logs are propagated
+    assert len(caplog.records) == 6
+    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
+        assert expected_message_and_level[0] == record.message
+        assert expected_message_and_level[1] == record.levelno
+
+
+@respx.mock
+async def test_actor_call_no_redirect_logs(
+    caplog: LogCaptureFixture,
+    mock_api: None,  # noqa: ARG001, fixture
+    propagate_stream_logs: None,  # noqa: ARG001, fixture
+) -> None:
+    logger_name = f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}'
+    run_client = ApifyClientAsync(token='mocked_token', api_url=_MOCKED_API_URL).actor(actor_id=_MOCKED_ACTOR_ID)
+
+    with caplog.at_level(logging.DEBUG, logger=logger_name):
+        await run_client.call(logger=None)
+
+    assert len(caplog.records) == 0
+
+
+@respx.mock
+async def test_actor_call_redirect_logs_to_custom_logger(
+    caplog: LogCaptureFixture,
+    mock_api: None,  # noqa: ARG001, fixture
+    propagate_stream_logs: None,  # noqa: ARG001, fixture
+) -> None:
+    """Test that logs are redirected correctly to the custom logger."""
+    logger_name = 'custom_logger'
+    logger = logging.getLogger(logger_name)
+    run_client = ApifyClientAsync(token='mocked_token', api_url=_MOCKED_API_URL).actor(actor_id=_MOCKED_ACTOR_ID)
+
+    with caplog.at_level(logging.DEBUG, logger=logger_name):
+        await run_client.call(logger=logger)
+
+    assert len(caplog.records) == 6
+    for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
+        assert expected_message_and_level[0] == record.message
+        assert expected_message_and_level[1] == record.levelno