From fbab73ac02991f421d11d327fa296c892dd2779e Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 18 Aug 2026 14:39:38 +0200 Subject: [PATCH 1/3] fix: Classify and retry failures while reading a streamed error body --- src/apify_client/http_clients/_impit.py | 42 ++++++++++++++++----- tests/unit/test_http_clients.py | 49 +++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 10 deletions(-) diff --git a/src/apify_client/http_clients/_impit.py b/src/apify_client/http_clients/_impit.py index 68353b1d..c0f1f6e3 100644 --- a/src/apify_client/http_clients/_impit.py +++ b/src/apify_client/http_clients/_impit.py @@ -4,6 +4,7 @@ import logging import random import time +from contextlib import suppress from datetime import timedelta from http import HTTPStatus from typing import TYPE_CHECKING, Any, TypeVar @@ -53,6 +54,13 @@ def _is_retryable_error(exc: Exception) -> bool: ) +def _stop_retrying_if_permanent(exc: Exception, *, stop_retrying: Callable[[], None]) -> None: + """Stop the retry loop when an exception is not a transient transport failure.""" + if not _is_retryable_error(exc): + logger.debug('Exception is not retryable', exc_info=exc) + stop_retrying() + + @docs_group('HTTP clients') class ImpitHttpClient(HttpClient): """Synchronous HTTP client for the Apify API built on top of [Impit](https://github.com/apify/impit). @@ -232,9 +240,7 @@ def _make_request( except Exception as exc: logger.debug('Request threw exception', exc_info=exc) - if not _is_retryable_error(exc): - logger.debug('Exception is not retryable', exc_info=exc) - stop_retrying() + _stop_retrying_if_permanent(exc, stop_retrying=stop_retrying) raise # Retry only server errors (5xx) and rate limits (429). @@ -246,8 +252,17 @@ def _make_request( 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 we can raise the error properly. - response.read() + # Read the response in case it is a stream, so we can raise the error properly. A read that fails is a + # transport failure like any other, so it goes through the same classification as a failed send. + try: + response.read() + except Exception as exc: + logger.debug('Reading the error response failed', exc_info=exc) + with suppress(Exception): + response.close() + _stop_retrying_if_permanent(exc, stop_retrying=stop_retrying) + raise + raise ApifyApiError(response, attempt, method=method) @staticmethod @@ -494,9 +509,7 @@ async def _make_request( except Exception as exc: logger.debug('Request threw exception', exc_info=exc) - if not _is_retryable_error(exc): - logger.debug('Exception is not retryable', exc_info=exc) - stop_retrying() + _stop_retrying_if_permanent(exc, stop_retrying=stop_retrying) raise # Retry only server errors (5xx) and rate limits (429). @@ -508,8 +521,17 @@ async def _make_request( 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 we can raise the error properly. - await response.aread() + # Read the response in case it is a stream, so we can raise the error properly. A read that fails is a + # transport failure like any other, so it goes through the same classification as a failed send. + try: + await response.aread() + except Exception as exc: + logger.debug('Reading the error response failed', exc_info=exc) + with suppress(Exception): + await response.aclose() + _stop_retrying_if_permanent(exc, stop_retrying=stop_retrying) + raise + raise ApifyApiError(response, attempt, method=method) @staticmethod diff --git a/tests/unit/test_http_clients.py b/tests/unit/test_http_clients.py index 34236887..b25f6978 100644 --- a/tests/unit/test_http_clients.py +++ b/tests/unit/test_http_clients.py @@ -323,6 +323,55 @@ def test_is_retryable_error() -> None: assert not _is_retryable_error(Exception('test')) +def test_error_response_read_failure_is_retried_and_closed() -> None: + """A failure while buffering a streamed error body is retried like a failed send, and the response is closed.""" + client = ImpitHttpClient(token='test_token', max_retries=1, min_delay_between_retries=timedelta(0)) + responses = [ + Mock(status_code=500, read=Mock(side_effect=impit.ReadError('truncated')), close=Mock()) for _ in range(2) + ] + request = Mock(side_effect=responses) + client._impit_client = Mock(request=request) + + with pytest.raises(impit.ReadError): + client.call(method='GET', url='https://api.test.com/endpoint', stream=True) + + assert request.call_count == 2 + for response in responses: + response.close.assert_called_once() + + +async def test_error_response_read_failure_is_retried_and_closed_async() -> None: + """The async client also retries a failed buffering of a streamed error body and closes the response.""" + client = ImpitHttpClientAsync(token='test_token', max_retries=1, min_delay_between_retries=timedelta(0)) + responses = [ + Mock(status_code=500, aread=AsyncMock(side_effect=impit.ReadError('truncated')), aclose=AsyncMock()) + for _ in range(2) + ] + request = AsyncMock(side_effect=responses) + client._impit_async_client = Mock(request=request) + + with pytest.raises(impit.ReadError): + await client.call(method='GET', url='https://api.test.com/endpoint', stream=True) + + assert request.call_count == 2 + for response in responses: + response.aclose.assert_awaited_once() + + +def test_non_retryable_error_response_read_failure_stops_retrying() -> None: + """A read failure that is not a transport error stops the retry loop instead of being retried.""" + client = ImpitHttpClient(token='test_token', min_delay_between_retries=timedelta(0)) + response = Mock(status_code=500, read=Mock(side_effect=ValueError('broken response')), close=Mock()) + request = Mock(return_value=response) + client._impit_client = Mock(request=request) + + with pytest.raises(ValueError, match='broken response'): + client.call(method='GET', url='https://api.test.com/endpoint', stream=True) + + request.assert_called_once() + response.close.assert_called_once() + + @pytest.fixture( params=[ pytest.param((GzipHttpCompressor(), 'gzip', gzip.decompress), id='gzip'), From 92cbd92eae8444cb92777657a56f6654dabc6ab6 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 18 Aug 2026 15:47:16 +0200 Subject: [PATCH 2/3] test: Cover read failures on the async and non-retryable status paths --- src/apify_client/http_clients/_impit.py | 8 +++---- tests/unit/test_http_clients.py | 30 ++++++++++++++++++++++++- 2 files changed, 33 insertions(+), 5 deletions(-) diff --git a/src/apify_client/http_clients/_impit.py b/src/apify_client/http_clients/_impit.py index c0f1f6e3..5d287804 100644 --- a/src/apify_client/http_clients/_impit.py +++ b/src/apify_client/http_clients/_impit.py @@ -252,8 +252,8 @@ def _make_request( 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 we can raise the error properly. A read that fails is a - # transport failure like any other, so it goes through the same classification as a failed send. + # Read the response in case it is a stream, so we can raise the error properly. A failed read goes through + # the same classification as a failed send. try: response.read() except Exception as exc: @@ -521,8 +521,8 @@ async def _make_request( 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 we can raise the error properly. A read that fails is a - # transport failure like any other, so it goes through the same classification as a failed send. + # Read the response in case it is a stream, so we can raise the error properly. A failed read goes through + # the same classification as a failed send. try: await response.aread() except Exception as exc: diff --git a/tests/unit/test_http_clients.py b/tests/unit/test_http_clients.py index b25f6978..10dddede 100644 --- a/tests/unit/test_http_clients.py +++ b/tests/unit/test_http_clients.py @@ -353,7 +353,7 @@ async def test_error_response_read_failure_is_retried_and_closed_async() -> None with pytest.raises(impit.ReadError): await client.call(method='GET', url='https://api.test.com/endpoint', stream=True) - assert request.call_count == 2 + assert request.await_count == 2 for response in responses: response.aclose.assert_awaited_once() @@ -372,6 +372,34 @@ def test_non_retryable_error_response_read_failure_stops_retrying() -> None: response.close.assert_called_once() +async def test_non_retryable_error_response_read_failure_stops_retrying_async() -> None: + """The async client also stops retrying when a read failure is not a transport error.""" + client = ImpitHttpClientAsync(token='test_token', min_delay_between_retries=timedelta(0)) + response = Mock(status_code=500, aread=AsyncMock(side_effect=ValueError('broken response')), aclose=AsyncMock()) + request = AsyncMock(return_value=response) + client._impit_async_client = Mock(request=request) + + with pytest.raises(ValueError, match='broken response'): + await client.call(method='GET', url='https://api.test.com/endpoint', stream=True) + + request.assert_awaited_once() + response.aclose.assert_awaited_once() + + +def test_error_response_read_failure_on_non_retryable_status_is_not_retried() -> None: + """A transient read failure on a status that is not retryable surfaces immediately instead of being retried.""" + client = ImpitHttpClient(token='test_token', min_delay_between_retries=timedelta(0)) + response = Mock(status_code=404, read=Mock(side_effect=impit.ReadError('truncated')), close=Mock()) + request = Mock(return_value=response) + client._impit_client = Mock(request=request) + + with pytest.raises(impit.ReadError): + client.call(method='GET', url='https://api.test.com/endpoint', stream=True) + + request.assert_called_once() + response.close.assert_called_once() + + @pytest.fixture( params=[ pytest.param((GzipHttpCompressor(), 'gzip', gzip.decompress), id='gzip'), From a810910aefa03f98b75ff9037111632a00dfd1c7 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 18 Aug 2026 18:19:41 +0200 Subject: [PATCH 3/3] refactor: Inline permanent-error classification at retry call sites --- src/apify_client/http_clients/_impit.py | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/src/apify_client/http_clients/_impit.py b/src/apify_client/http_clients/_impit.py index 5d287804..dbbe01ab 100644 --- a/src/apify_client/http_clients/_impit.py +++ b/src/apify_client/http_clients/_impit.py @@ -54,13 +54,6 @@ def _is_retryable_error(exc: Exception) -> bool: ) -def _stop_retrying_if_permanent(exc: Exception, *, stop_retrying: Callable[[], None]) -> None: - """Stop the retry loop when an exception is not a transient transport failure.""" - if not _is_retryable_error(exc): - logger.debug('Exception is not retryable', exc_info=exc) - stop_retrying() - - @docs_group('HTTP clients') class ImpitHttpClient(HttpClient): """Synchronous HTTP client for the Apify API built on top of [Impit](https://github.com/apify/impit). @@ -240,7 +233,9 @@ def _make_request( except Exception as exc: logger.debug('Request threw exception', exc_info=exc) - _stop_retrying_if_permanent(exc, stop_retrying=stop_retrying) + if not _is_retryable_error(exc): + logger.debug('Exception is not retryable', exc_info=exc) + stop_retrying() raise # Retry only server errors (5xx) and rate limits (429). @@ -260,7 +255,9 @@ def _make_request( logger.debug('Reading the error response failed', exc_info=exc) with suppress(Exception): response.close() - _stop_retrying_if_permanent(exc, stop_retrying=stop_retrying) + if not _is_retryable_error(exc): + logger.debug('Exception is not retryable', exc_info=exc) + stop_retrying() raise raise ApifyApiError(response, attempt, method=method) @@ -509,7 +506,9 @@ async def _make_request( except Exception as exc: logger.debug('Request threw exception', exc_info=exc) - _stop_retrying_if_permanent(exc, stop_retrying=stop_retrying) + if not _is_retryable_error(exc): + logger.debug('Exception is not retryable', exc_info=exc) + stop_retrying() raise # Retry only server errors (5xx) and rate limits (429). @@ -529,7 +528,9 @@ async def _make_request( logger.debug('Reading the error response failed', exc_info=exc) with suppress(Exception): await response.aclose() - _stop_retrying_if_permanent(exc, stop_retrying=stop_retrying) + if not _is_retryable_error(exc): + logger.debug('Exception is not retryable', exc_info=exc) + stop_retrying() raise raise ApifyApiError(response, attempt, method=method)