diff --git a/docs/02_concepts/11_timeouts.mdx b/docs/02_concepts/11_timeouts.mdx index 55b937de..3bd0748e 100644 --- a/docs/02_concepts/11_timeouts.mdx +++ b/docs/02_concepts/11_timeouts.mdx @@ -25,7 +25,7 @@ Every client method has a pre-assigned tier that matches the expected duration o ## Configuring default timeouts -You can override the default values for each tier in the `ApifyClient` or `ApifyClientAsync` constructor. The `timeout_max` parameter sets an upper cap on the timeout for any individual API request, limiting exponential growth during retries. +You can override the default values for each tier in the `ApifyClient` or `ApifyClientAsync` constructor. The `timeout_max` parameter sets an upper cap on the timeout for any individual API request, limiting exponential growth during retries. It caps tier and per-call timeouts as well, so raise it whenever you need a timeout longer than the default 360 seconds. A tier configured above the cap is capped too, and the client logs a warning to make the cut-off visible. @@ -57,6 +57,8 @@ client.dataset('id').list_items(timeout='long') client.dataset('id').list_items(timeout='no_timeout') ``` +A `timedelta` longer than `timeout_max` is capped at `timeout_max`, and the client logs a warning. To let such a call use its full timeout, raise `timeout_max` in the client constructor. + ## Interaction with retries Timeouts work together with the [retry system](/api/client/python/docs/concepts/retries). When a request times out, it counts as a failed attempt and triggers a retry (up to `max_retries`). The timeout applies to each individual attempt, not the total time across all retries. diff --git a/src/apify_client/_apify_client.py b/src/apify_client/_apify_client.py index 391906c0..6ca9413d 100644 --- a/src/apify_client/_apify_client.py +++ b/src/apify_client/_apify_client.py @@ -146,7 +146,7 @@ def __init__( timeout_short: Default timeout for short-duration API operations (simple CRUD operations, ...). timeout_medium: Default timeout for medium-duration API operations (batch operations, listing, ...). timeout_long: Default timeout for long-duration API operations (long-polling, streaming, ...). - timeout_max: Maximum timeout cap for exponential timeout growth across retries. + timeout_max: Maximum timeout cap for any single request attempt, including tier and per-call timeouts. headers: Additional HTTP headers to include in all API requests. compression: Compression algorithm for request bodies. Pass a string literal to select an algorithm, or an `HttpCompressor` instance for finer-grained control. @@ -508,7 +508,7 @@ def __init__( timeout_short: Default timeout for short-duration API operations (simple CRUD operations, ...). timeout_medium: Default timeout for medium-duration API operations (batch operations, listing, ...). timeout_long: Default timeout for long-duration API operations (long-polling, streaming, ...). - timeout_max: Maximum timeout cap for exponential timeout growth across retries. + timeout_max: Maximum timeout cap for any single request attempt, including tier and per-call timeouts. headers: Additional HTTP headers to include in all API requests. compression: Compression algorithm for request bodies. Pass a string literal to select an algorithm, or an `HttpCompressor` instance for finer-grained control. diff --git a/src/apify_client/_logging.py b/src/apify_client/_logging.py index 989d0a36..cdd2374b 100644 --- a/src/apify_client/_logging.py +++ b/src/apify_client/_logging.py @@ -3,6 +3,7 @@ import functools import inspect import logging +import threading from contextvars import ContextVar from typing import TYPE_CHECKING, Any, NamedTuple @@ -42,6 +43,38 @@ class LogContext(NamedTuple): ) +class LoggerOnce: + """Emits each log message at most once, keyed by an explicit string. + + Useful for diagnostic warnings that would otherwise spam the log when the same condition recurs (per-request + misconfiguration warnings, repeated fallback paths, etc.). Deduplication scope follows the lifetime of the + instance - a module-level instance gives process-wide dedup; an attribute on a class gives per-instance dedup. + + Safe to call from multiple threads, since the sync client can be used from several threads at once. + """ + + def __init__(self, logger: logging.Logger) -> None: + self._logger = logger + self._seen: set[str] = set() + self._lock = threading.Lock() + + def log(self, message: str, *, key: str, level: int = logging.INFO) -> None: + """Log `message` at `level` the first time `key` is seen on this instance; later calls are no-ops. + + Args: + message: The message to log. + key: Deduplication key. Two calls with the same key emit at most once. + level: Standard `logging` level (e.g. `logging.WARNING`). Defaults to `logging.INFO`. + """ + # The check and the insert have to be atomic, otherwise two threads racing on the same key both emit. + with self._lock: + if key in self._seen: + return + self._seen.add(key) + + self._logger.log(level, message) + + class WithLogDetailsClient(type): """Metaclass that wraps public methods to inject client details into log context.""" diff --git a/src/apify_client/http_clients/_base.py b/src/apify_client/http_clients/_base.py index 7645228f..620c79be 100644 --- a/src/apify_client/http_clients/_base.py +++ b/src/apify_client/http_clients/_base.py @@ -1,6 +1,7 @@ from __future__ import annotations import json as jsonlib +import logging import os import sys from abc import ABC, abstractmethod @@ -18,6 +19,7 @@ DEFAULT_TIMEOUT_SHORT, ) from apify_client._docs import docs_group +from apify_client._logging import LoggerOnce, logger_name from apify_client._statistics import ClientStatistics from apify_client._utils.time import to_seconds from apify_client.http_compressors._gzip import GzipHttpCompressor @@ -28,6 +30,9 @@ from apify_client.http_compressors._base import HttpCompressor from apify_client.types import JsonSerializable, Timeout +logger = logging.getLogger(logger_name) +logger_once = LoggerOnce(logger) + @docs_group('HTTP clients') @runtime_checkable @@ -110,7 +115,7 @@ def __init__( timeout_short: Default timeout for short-duration API operations (simple CRUD operations, ...). timeout_medium: Default timeout for medium-duration API operations (batch operations, listing, ...). timeout_long: Default timeout for long-duration API operations (long-polling, streaming, ...). - timeout_max: Maximum timeout cap for exponential timeout growth across retries. + timeout_max: Maximum timeout cap for any single request attempt, including tier and per-call timeouts. max_retries: Maximum number of retries for failed requests. min_delay_between_retries: Minimum delay between retries. statistics: Statistics tracker for API calls. Created automatically if not provided. @@ -198,7 +203,8 @@ def _compute_timeout(self, timeout: Timeout, *, attempt: int) -> int | float | N """Resolve a timeout tier and compute the timeout for a request attempt with exponential increase. For `no_timeout`, returns `None` to indicate no timeout. For tier literals and explicit `timedelta` values, - doubles the timeout with each attempt but caps at `timeout_max`. + doubles the timeout with each attempt but caps at `timeout_max`. A base timeout above `timeout_max` is + capped too, which warns once per timeout kind since the requested value does not take effect in full. Args: timeout: The timeout specification to resolve (tier literal or explicit `timedelta`). @@ -219,6 +225,16 @@ def _compute_timeout(self, timeout: Timeout, *, attempt: int) -> int | float | N else: resolved = timeout + if resolved > self._timeout_max: + # Keyed per timeout kind, so retries and repeated calls do not spam the log with the same warning. + logger_once.log( + f'The requested timeout of {to_seconds(resolved)}s exceeds `timeout_max` ' + f'({to_seconds(self._timeout_max)}s) and is capped at it. Raise `timeout_max` on the client ' + 'to allow longer request timeouts.', + key=f'timeout-capped-{timeout if isinstance(timeout, str) else "explicit"}', + level=logging.WARNING, + ) + new_timeout = min(resolved * (2 ** (attempt - 1)), self._timeout_max) return to_seconds(new_timeout) @@ -312,8 +328,8 @@ def call( json: JSON-serializable data for the request body. Cannot be used together with data. stream: Whether to stream the response body. timeout: Timeout for the API HTTP request. Use `short`, `medium`, or `long` tier literals for - preconfigured timeouts. A `timedelta` overrides it for this call, and `no_timeout` disables - the timeout entirely. + preconfigured timeouts. A `timedelta` overrides it for this call (capped at `timeout_max`), and + `no_timeout` disables the timeout entirely. Returns: The HTTP response object. @@ -356,8 +372,8 @@ async def call( json: JSON-serializable data for the request body. Cannot be used together with data. stream: Whether to stream the response body. timeout: Timeout for the API HTTP request. Use `short`, `medium`, or `long` tier literals for - preconfigured timeouts. A `timedelta` overrides it for this call, and `no_timeout` disables - the timeout entirely. + preconfigured timeouts. A `timedelta` overrides it for this call (capped at `timeout_max`), and + `no_timeout` disables the timeout entirely. Returns: The HTTP response object. diff --git a/src/apify_client/http_clients/_impit.py b/src/apify_client/http_clients/_impit.py index c3ef7212..c71d2686 100644 --- a/src/apify_client/http_clients/_impit.py +++ b/src/apify_client/http_clients/_impit.py @@ -83,7 +83,7 @@ def __init__( timeout_short: Default timeout for short-duration API operations (simple CRUD operations, ...). timeout_medium: Default timeout for medium-duration API operations (batch operations, listing, ...). timeout_long: Default timeout for long-duration API operations (long-polling, streaming, ...). - timeout_max: Maximum timeout cap for exponential timeout growth across retries. + timeout_max: Maximum timeout cap for any single request attempt, including tier and per-call timeouts. max_retries: Maximum number of retry attempts for failed requests. min_delay_between_retries: Minimum delay between retries (increases exponentially with each attempt). statistics: Statistics tracker for API calls. Created automatically if not provided. @@ -130,8 +130,8 @@ def call( json: JSON-serializable data for the request body. Cannot be used together with data. stream: Whether to stream the response body. timeout: Timeout for the API HTTP request. Use `short`, `medium`, or `long` tier literals for - preconfigured timeouts. A `timedelta` overrides it for this call, and `no_timeout` disables - the timeout entirely. + preconfigured timeouts. A `timedelta` overrides it for this call (capped at `timeout_max`), and + `no_timeout` disables the timeout entirely. Returns: The HTTP response object. @@ -332,7 +332,7 @@ def __init__( timeout_short: Default timeout for short-duration API operations (simple CRUD operations, ...). timeout_medium: Default timeout for medium-duration API operations (batch operations, listing, ...). timeout_long: Default timeout for long-duration API operations (long-polling, streaming, ...). - timeout_max: Maximum timeout cap for exponential timeout growth across retries. + timeout_max: Maximum timeout cap for any single request attempt, including tier and per-call timeouts. max_retries: Maximum number of retry attempts for failed requests. min_delay_between_retries: Minimum delay between retries (increases exponentially with each attempt). statistics: Statistics tracker for API calls. Created automatically if not provided. @@ -379,8 +379,8 @@ async def call( json: JSON-serializable data for the request body. Cannot be used together with data. stream: Whether to stream the response body. timeout: Timeout for the API HTTP request. Use `short`, `medium`, or `long` tier literals for - preconfigured timeouts. A `timedelta` overrides it for this call, and `no_timeout` disables - the timeout entirely. + preconfigured timeouts. A `timedelta` overrides it for this call (capped at `timeout_max`), and + `no_timeout` disables the timeout entirely. Returns: The HTTP response object. diff --git a/tests/unit/test_client_timeouts.py b/tests/unit/test_client_timeouts.py index 2e87279e..03ddf36b 100644 --- a/tests/unit/test_client_timeouts.py +++ b/tests/unit/test_client_timeouts.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging from datetime import timedelta from typing import TYPE_CHECKING, Any from unittest.mock import Mock @@ -7,11 +8,15 @@ import pytest from impit import HTTPError, Response, TimeoutException +from apify_client._logging import LoggerOnce, logger_name from apify_client.http_clients import ImpitHttpClient, ImpitHttpClientAsync +from apify_client.http_clients import _base as http_client_base if TYPE_CHECKING: from collections.abc import Iterator + from _pytest.logging import LogCaptureFixture + class EndOfTestError(Exception): """Custom exception that is raised after the relevant part of the code is executed to stop the test.""" @@ -34,6 +39,12 @@ async def mock_request_async(*args: Any, **kwargs: Any) -> None: monkeypatch.undo() +@pytest.fixture +def fresh_logger_once(monkeypatch: pytest.MonkeyPatch) -> None: + """Replace the module-level `LoggerOnce`, whose dedup state would otherwise leak between tests.""" + monkeypatch.setattr(http_client_base, 'logger_once', LoggerOnce(http_client_base.logger)) + + def test_no_timeout_passes_large_value_to_impit_sync(patch_request: list) -> None: """Test that `no_timeout` passes a large timeout to impit to effectively disable the timeout.""" client = ImpitHttpClient(timeout_short=timedelta(seconds=10)) @@ -149,6 +160,47 @@ def test_compute_timeout_no_timeout_returns_none() -> None: assert client._compute_timeout('no_timeout', attempt=1) is None +@pytest.mark.usefixtures('fresh_logger_once') +def test_compute_timeout_explicit_timedelta_above_max_warns(caplog: LogCaptureFixture) -> None: + """Test an explicit timedelta larger than timeout_max is capped, and the cut-off is logged once.""" + client = ImpitHttpClient(timeout_max=timedelta(seconds=360)) + + with caplog.at_level(logging.WARNING, logger=logger_name): + assert client._compute_timeout(timedelta(minutes=30), attempt=1) == 360.0 + # Retries and later calls recompute the timeout, which must not repeat the warning. + assert client._compute_timeout(timedelta(minutes=30), attempt=2) == 360.0 + assert client._compute_timeout(timedelta(minutes=45), attempt=1) == 360.0 + + assert len(caplog.records) == 1 + assert '1800.0s exceeds `timeout_max` (360.0s)' in caplog.records[0].message + + +@pytest.mark.usefixtures('fresh_logger_once') +def test_compute_timeout_tier_above_max_warns(caplog: LogCaptureFixture) -> None: + """Test a tier configured larger than timeout_max is capped, and the cut-off is logged too.""" + client = ImpitHttpClient(timeout_long=timedelta(seconds=600), timeout_max=timedelta(seconds=360)) + + with caplog.at_level(logging.WARNING, logger=logger_name): + assert client._compute_timeout('long', attempt=1) == 360.0 + + assert len(caplog.records) == 1 + assert '600.0s exceeds `timeout_max` (360.0s)' in caplog.records[0].message + + +@pytest.mark.usefixtures('fresh_logger_once') +def test_compute_timeout_within_max_does_not_warn(caplog: LogCaptureFixture) -> None: + """Test a base timeout within timeout_max is used as-is, without a warning.""" + client = ImpitHttpClient(timeout_long=timedelta(seconds=300), timeout_max=timedelta(seconds=360)) + + with caplog.at_level(logging.WARNING, logger=logger_name): + assert client._compute_timeout(timedelta(seconds=120), attempt=1) == 120.0 + assert client._compute_timeout('long', attempt=1) == 300.0 + # Growth capped at `timeout_max` on a retry is expected, so it is not warned about. + assert client._compute_timeout('long', attempt=2) == 360.0 + + assert caplog.records == [] + + async def test_dynamic_timeout_async_client(monkeypatch: pytest.MonkeyPatch) -> None: """Tests timeout values for request with retriable errors. diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index a5148696..4749c10c 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -13,7 +13,7 @@ from werkzeug import Request, Response from apify_client import ApifyClient, ApifyClientAsync -from apify_client._logging import RedirectLogFormatter +from apify_client._logging import LoggerOnce, RedirectLogFormatter from apify_client._status_message_watcher import StatusMessageWatcherBase from apify_client._streamed_log import StreamedLog, StreamedLogAsync, StreamedLogBase @@ -997,3 +997,65 @@ def generate_logs() -> Iterator[bytes]: assert not error_records, f'async task logged an error on stream timeout: {[r.message for r in error_records]}' # The line received before the timeout must still have been redirected. assert any('ACTOR: still running' in record.message for record in caplog.records) + + +def test_logger_once_logs_the_first_call(caplog: LogCaptureFixture) -> None: + """Test the first call with a given key is logged.""" + logger = logging.getLogger('apify_client.tests.log_once_first') + logger_once = LoggerOnce(logger) + + with caplog.at_level(logging.INFO, logger=logger.name): + logger_once.log('first', key='k1') + + assert [record.getMessage() for record in caplog.records] == ['first'] + + +def test_logger_once_suppresses_a_repeated_key(caplog: LogCaptureFixture) -> None: + """Test a second call with an already seen key is a no-op.""" + logger = logging.getLogger('apify_client.tests.log_once_repeated') + logger_once = LoggerOnce(logger) + + with caplog.at_level(logging.INFO, logger=logger.name): + logger_once.log('first', key='k1') + logger_once.log('second', key='k1') + + assert [record.getMessage() for record in caplog.records] == ['first'] + + +def test_logger_once_logs_distinct_keys(caplog: LogCaptureFixture) -> None: + """Test each distinct key is logged once on its own.""" + logger = logging.getLogger('apify_client.tests.log_once_distinct') + logger_once = LoggerOnce(logger) + + with caplog.at_level(logging.INFO, logger=logger.name): + logger_once.log('msg-a', key='k1') + logger_once.log('msg-b', key='k2') + + assert [record.getMessage() for record in caplog.records] == ['msg-a', 'msg-b'] + + +def test_logger_once_instances_keep_independent_state(caplog: LogCaptureFixture) -> None: + """Test dedup state is per instance, so two instances sharing a logger both emit the same key.""" + logger = logging.getLogger('apify_client.tests.log_once_independent') + logger_once_a = LoggerOnce(logger) + logger_once_b = LoggerOnce(logger) + + with caplog.at_level(logging.INFO, logger=logger.name): + logger_once_a.log('from-a', key='k1') + logger_once_b.log('from-b', key='k1') + + assert [record.getMessage() for record in caplog.records] == ['from-a', 'from-b'] + + +def test_logger_once_logs_at_the_requested_level(caplog: LogCaptureFixture) -> None: + """Test the level defaults to info and is honored when passed explicitly.""" + logger = logging.getLogger('apify_client.tests.log_once_levels') + logger_once = LoggerOnce(logger) + + with caplog.at_level(logging.DEBUG, logger=logger.name): + logger_once.log('default-msg', key='k_default') + logger_once.log('warn-msg', key='k_warn', level=logging.WARNING) + + levels = {record.getMessage(): record.levelno for record in caplog.records} + assert levels['default-msg'] == logging.INFO + assert levels['warn-msg'] == logging.WARNING