refactor: Deduplicate the response-status retry policy into the HTTP client base

Author: vdusek

src/apify_client/_apify_client.py

@@ -233,7 +233,13 @@ def with_custom_http_client(
 
         ```python
         from apify_client import ApifyClient
-        from my_http_client import MyHttpClient
+        from apify_client.http_clients import HttpClient, HttpResponse
+
+
+        class MyHttpClient(HttpClient):
+            def send_request(self, *, method, url, headers, content, timeout, stream) -> HttpResponse:
+                ...
+
 
         client = ApifyClient.with_custom_http_client(
             token='MY-APIFY-TOKEN',
@@ -591,7 +597,13 @@ def with_custom_http_client(
 
         ```python
         from apify_client import ApifyClientAsync
-        from my_http_client import MyHttpClientAsync
+        from apify_client.http_clients import HttpClientAsync, HttpResponse
+
+
+        class MyHttpClientAsync(HttpClientAsync):
+            async def send_request(self, *, method, url, headers, content, timeout, stream) -> HttpResponse:
+                ...
+
 
         client = ApifyClientAsync.with_custom_http_client(
             token='MY-APIFY-TOKEN',

src/apify_client/http_clients/_base.py

@@ -101,12 +101,11 @@ def aiter_bytes(self) -> AsyncIterator[bytes]:
 class HttpClientBase:
     """Shared configuration and utilities for HTTP clients.
 
-    Provides common functionality for both sync and async HTTP clients including:
-    header construction, parameter parsing, request body preparation, URL building,
-    and timeout calculation.
+    Provides common functionality for both sync and async HTTP clients including: header construction, parameter
+    parsing, request body preparation, URL building, timeout calculation, and error classification.
 
-    Subclasses should call `super().__init__()` to initialize shared configuration.
-    The helper methods are then available for use in the `call()` implementation.
+    Subclasses should call `super().__init__()` to initialize shared configuration. The helpers are then used by the
+    inherited `call`, and stay available to a client that replaces it.
     """
 
     def __init__(
@@ -365,15 +364,44 @@ def _handle_request_exception(self, exc: Exception, *, stop_retrying: Callable[[
             logger.debug('Exception is not retryable', exc_info=exc)
             stop_retrying()
 
+    def _handle_response_status(
+        self,
+        response: HttpResponse,
+        *,
+        attempt: int,
+        stop_retrying: Callable[[], None],
+    ) -> bool:
+        """Record the response status and stop retrying unless it is a server error or a rate limit.
+
+        Returns whether the response is a success, so the caller can hand it back instead of raising.
+        """
+        if response.status_code < HTTPStatus.MULTIPLE_CHOICES:
+            logger.debug('Request successful', extra={'status_code': response.status_code})
+            return True
+
+        if response.status_code == HTTPStatus.TOO_MANY_REQUESTS:
+            self._statistics.add_rate_limit_error(attempt)
+
+        logger.debug('Request unsuccessful', extra={'status_code': response.status_code})
+        if (
+            response.status_code < HTTPStatus.INTERNAL_SERVER_ERROR
+            and response.status_code != HTTPStatus.TOO_MANY_REQUESTS
+        ):
+            logger.debug('Status code is not retryable', extra={'status_code': response.status_code})
+            stop_retrying()
+
+        return False
+
 
 @docs_group('HTTP clients')
 class HttpClient(HttpClientBase):
     """Base class for synchronous HTTP clients used by `ApifyClient`.
 
     Concrete clients inherit its request preparation, retry, and error handling, and implement the transport,
     error-classification, and lifecycle hooks. Only `send_request` has to be implemented, the other hooks have
-    defaults. Replacing `call` itself also remains supported, and then none of the hooks are used. Helper methods
-    from `HttpClientBase` remain available for request preparation, URL building, and parameter parsing.
+    defaults. Replacing `call` itself also remains supported, and then the transport and retry-classification hooks
+    are bypassed. Helper methods from `HttpClientBase` remain available for request preparation, URL building, and
+    parameter parsing.
 
     The client's default headers from `self._headers` have to go out with every request, otherwise the
     `Authorization` header never reaches the API. The inherited `call` merges them into the per-request headers
@@ -556,21 +584,9 @@ def _make_request(
             self._handle_request_exception(exc, stop_retrying=stop_retrying)
             raise
 
-        if response.status_code < HTTPStatus.MULTIPLE_CHOICES:
-            logger.debug('Request successful', extra={'status_code': response.status_code})
+        if self._handle_response_status(response, attempt=attempt, stop_retrying=stop_retrying):
             return response
 
-        if response.status_code == HTTPStatus.TOO_MANY_REQUESTS:
-            self._statistics.add_rate_limit_error(attempt)
-
-        logger.debug('Request unsuccessful', extra={'status_code': response.status_code})
-        if (
-            response.status_code < HTTPStatus.INTERNAL_SERVER_ERROR
-            and response.status_code != HTTPStatus.TOO_MANY_REQUESTS
-        ):
-            logger.debug('Status code is not retryable', extra={'status_code': response.status_code})
-            stop_retrying()
-
         # Read the response in case it is a stream, so the error can be raised properly.
         response.read()
         raise ApifyApiError(response, attempt, method=method)
@@ -773,21 +789,9 @@ async def _make_request(
             self._handle_request_exception(exc, stop_retrying=stop_retrying)
             raise
 
-        if response.status_code < HTTPStatus.MULTIPLE_CHOICES:
-            logger.debug('Request successful', extra={'status_code': response.status_code})
+        if self._handle_response_status(response, attempt=attempt, stop_retrying=stop_retrying):
             return response
 
-        if response.status_code == HTTPStatus.TOO_MANY_REQUESTS:
-            self._statistics.add_rate_limit_error(attempt)
-
-        logger.debug('Request unsuccessful', extra={'status_code': response.status_code})
-        if (
-            response.status_code < HTTPStatus.INTERNAL_SERVER_ERROR
-            and response.status_code != HTTPStatus.TOO_MANY_REQUESTS
-        ):
-            logger.debug('Status code is not retryable', extra={'status_code': response.status_code})
-            stop_retrying()
-
         # Read the response in case it is a stream, so the error can be raised properly.
         await response.aread()
         raise ApifyApiError(response, attempt, method=method)

src/apify_client/http_clients/_impit.py

@@ -79,6 +79,12 @@ def __init__(
     def is_timeout_error(self, exc: Exception) -> bool:
         return super().is_timeout_error(exc) or isinstance(exc, impit.TimeoutException)
 
+    @override
+    def is_retryable_transport_error(self, exc: Exception) -> bool:
+        # All errors from Impit's own hierarchy count as transient - they represent transport-level failures
+        # (network issues, timeouts, protocol errors, body decoding errors) that are typically transient.
+        return isinstance(exc, impit.HTTPError)
+
     @override
     def close(self) -> None:
         """Release resources owned by this client.
@@ -113,12 +119,6 @@ def send_request(
             stream=stream,
         )
 
-    @override
-    def is_retryable_transport_error(self, exc: Exception) -> bool:
-        # All errors from Impit's own hierarchy count as transient - they represent transport-level failures
-        # (network issues, timeouts, protocol errors, body decoding errors) that are typically transient.
-        return isinstance(exc, impit.HTTPError)
-
 
 @docs_group('HTTP clients')
 class ImpitHttpClientAsync(HttpClientAsync):
@@ -176,6 +176,12 @@ def __init__(
     def is_timeout_error(self, exc: Exception) -> bool:
         return super().is_timeout_error(exc) or isinstance(exc, impit.TimeoutException)
 
+    @override
+    def is_retryable_transport_error(self, exc: Exception) -> bool:
+        # All errors from Impit's own hierarchy count as transient - they represent transport-level failures
+        # (network issues, timeouts, protocol errors, body decoding errors) that are typically transient.
+        return isinstance(exc, impit.HTTPError)
+
     @override
     async def aclose(self) -> None:
         """Release resources owned by this client.
@@ -205,9 +211,3 @@ async def send_request(
             timeout=impit_timeout,
             stream=stream,
         )
-
-    @override
-    def is_retryable_transport_error(self, exc: Exception) -> bool:
-        # All errors from Impit's own hierarchy count as transient - they represent transport-level failures
-        # (network issues, timeouts, protocol errors, body decoding errors) that are typically transient.
-        return isinstance(exc, impit.HTTPError)

tests/unit/test_client_timeouts.py

@@ -93,7 +93,7 @@ async def test_timeout_resolves_for_async_clients(
 
 
 def test_compute_timeout_with_timedelta(http_client_class: type[HttpClient]) -> None:
-    """Concrete timedeltas double per attempt and are capped at the configured maximum."""
+    """Concrete timedeltas double per attempt, are capped at the maximum, and `no_timeout` stays unbounded."""
     client = http_client_class(timeout_max=timedelta(seconds=20))
 
     assert client._compute_timeout(timedelta(seconds=5), attempt=1) == 5.0

tests/unit/test_http_clients.py

@@ -72,7 +72,7 @@ async def test_call_only_http_client_async_keeps_working() -> None:
         assert not client.is_retryable_transport_error(TimeoutError('test'))
 
 
-def test_retry_with_exp_backoff(http_client_class: type[HttpClient]) -> None:
+def test_retry_with_exp_backoff() -> None:
     attempt_counter = 0
 
     class RetryableError(Exception):
@@ -101,7 +101,7 @@ def bails_on_third_attempt(stop_retrying: Callable, attempt: int) -> Any:
 
     # Returns the correct result after the correct time (should take 100 + 200 + 400 + 800 = 1500 ms)
     start = time.time()
-    result = http_client_class._retry_with_exp_backoff(
+    result = HttpClient._retry_with_exp_backoff(
         returns_on_fifth_attempt, backoff_base=timedelta(milliseconds=100), backoff_factor=2, random_factor=0
     )
     elapsed_time_seconds = time.time() - start
@@ -113,21 +113,19 @@ def bails_on_third_attempt(stop_retrying: Callable, attempt: int) -> Any:
     # Stops retrying when failed for max_retries times
     attempt_counter = 0
     with pytest.raises(RetryableError):
-        http_client_class._retry_with_exp_backoff(
+        HttpClient._retry_with_exp_backoff(
             returns_on_fifth_attempt, max_retries=3, backoff_base=timedelta(milliseconds=1)
         )
     assert attempt_counter == 4
 
     # Bails when the bail function is called
     attempt_counter = 0
     with pytest.raises(NonRetryableError):
-        http_client_class._retry_with_exp_backoff(bails_on_third_attempt, backoff_base=timedelta(milliseconds=1))
+        HttpClient._retry_with_exp_backoff(bails_on_third_attempt, backoff_base=timedelta(milliseconds=1))
     assert attempt_counter == 3
 
 
-async def test_retry_with_exp_backoff_async(
-    http_client_async_class: type[HttpClientAsync],
-) -> None:
+async def test_retry_with_exp_backoff_async() -> None:
     attempt_counter = 0
 
     class RetryableError(Exception):
@@ -156,7 +154,7 @@ async def bails_on_third_attempt(stop_retrying: Callable, attempt: int) -> Any:
 
     # Returns the correct result after the correct time (should take 100 + 200 + 400 + 800 = 1500 ms)
     start = time.time()
-    result = await http_client_async_class._retry_with_exp_backoff(
+    result = await HttpClientAsync._retry_with_exp_backoff(
         returns_on_fifth_attempt, backoff_base=timedelta(milliseconds=100), backoff_factor=2, random_factor=0
     )
     elapsed_time_seconds = time.time() - start
@@ -168,17 +166,15 @@ async def bails_on_third_attempt(stop_retrying: Callable, attempt: int) -> Any:
     # Stops retrying when failed for max_retries times
     attempt_counter = 0
     with pytest.raises(RetryableError):
-        await http_client_async_class._retry_with_exp_backoff(
+        await HttpClientAsync._retry_with_exp_backoff(
             returns_on_fifth_attempt, max_retries=3, backoff_base=timedelta(milliseconds=1)
         )
     assert attempt_counter == 4
 
     # Bails when the bail function is called
     attempt_counter = 0
     with pytest.raises(NonRetryableError):
-        await http_client_async_class._retry_with_exp_backoff(
-            bails_on_third_attempt, backoff_base=timedelta(milliseconds=1)
-        )
+        await HttpClientAsync._retry_with_exp_backoff(bails_on_third_attempt, backoff_base=timedelta(milliseconds=1))
     assert attempt_counter == 3
 
 
@@ -253,15 +249,15 @@ def test_merge_headers(
 
 
 def test_http_client_creates_sync_impit_client() -> None:
-    """The synchronous adapter creates the underlying Impit client and releases it on close."""
+    """The synchronous adapter creates the underlying Impit client and closes it through the close hook."""
     client = ImpitHttpClient(token='test_token_123')
 
     assert isinstance(client._impit_client, impit.Client)
     client.close()
 
 
 async def test_http_client_async_creates_async_impit_client() -> None:
-    """The asynchronous adapter creates the underlying Impit client and releases it on close."""
+    """The asynchronous adapter creates the underlying Impit client and closes it through the close hook."""
     client = ImpitHttpClientAsync(token='test_token_123')
 
     assert isinstance(client._impit_async_client, impit.AsyncClient)

tests/unit/test_logging.py

@@ -877,7 +877,7 @@ def test_streamed_log_sync_does_not_leak_exception_on_stream_timeout(
     http_client_class: type[HttpClient],
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """The streaming thread ends quietly when either transport times out while reading the log stream."""
+    """The streaming thread ends quietly when the transport times out while reading the log stream."""
     monkeypatch.setattr(StreamedLog, '_stream_timeout', timedelta(seconds=1))
 
     release_server = threading.Event()
@@ -970,7 +970,7 @@ async def test_streamed_log_async_does_not_error_on_stream_timeout(
     http_client_async_class: type[HttpClientAsync],
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """The async streaming task treats either transport's stream timeout as an expected terminal condition."""
+    """The async streaming task treats a transport stream timeout as an expected terminal condition."""
     monkeypatch.setattr(StreamedLogAsync, '_stream_timeout', timedelta(seconds=1))
 
     release_server = threading.Event()

tests/unit/test_pluggable_http_client.py

@@ -6,6 +6,7 @@
 from datetime import timedelta
 from http.client import HTTPConnection
 from typing import TYPE_CHECKING, Any
+from unittest.mock import AsyncMock, Mock
 from urllib.parse import urlsplit
 
 import impit
@@ -257,13 +258,13 @@ async def test_fake_response_async_methods() -> None:
 
 def test_http_client_without_transport_fails_loudly() -> None:
     """The sync base carries hook defaults, but its inherited `call` still needs a transport implementation."""
-    with pytest.raises(NotImplementedError):
+    with pytest.raises(NotImplementedError, match='Implement `send_request`'):
         HttpClient().call(method='GET', url='https://example.com')
 
 
 async def test_http_client_async_without_transport_fails_loudly() -> None:
     """The async base carries hook defaults, but its inherited `call` still needs a transport implementation."""
-    with pytest.raises(NotImplementedError):
+    with pytest.raises(NotImplementedError, match='Implement `send_request`'):
         await HttpClientAsync().call(method='GET', url='https://example.com')
 
 
@@ -725,3 +726,29 @@ def handler(_request: Request) -> Response:
 
     assert exc_info.value.status_code == 404
     assert request_count == 1
+
+
+def test_hooks_only_client_does_not_retry_an_unclassified_transport_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The default `is_retryable_transport_error` classifies nothing, so a transport failure ends the call at once."""
+    client = StdlibHttpClient(token='hook_token', min_delay_between_retries=timedelta(milliseconds=1))
+    send_request = Mock(side_effect=OSError('connection reset'))
+    monkeypatch.setattr(client, 'send_request', send_request)
+
+    with pytest.raises(OSError, match='connection reset'):
+        client.call(method='GET', url='https://example.com')
+
+    send_request.assert_called_once()
+
+
+async def test_hooks_only_client_async_does_not_retry_an_unclassified_transport_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The asynchronous pipeline gives up on the first transport failure the client does not classify either."""
+    client = StdlibHttpClientAsync(token='hook_token', min_delay_between_retries=timedelta(milliseconds=1))
+    send_request = AsyncMock(side_effect=OSError('connection reset'))
+    monkeypatch.setattr(client, 'send_request', send_request)
+
+    with pytest.raises(OSError, match='connection reset'):
+        await client.call(method='GET', url='https://example.com')
+
+    send_request.assert_awaited_once()