diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py index 866d6b7d2..52cabaa46 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py @@ -27,7 +27,8 @@ class HttpResponseFilter: Filter to select a response based on its HTTP status code, error message or a predicate. If a response matches the filter, the response action, failure_type, and error message are returned as an ErrorResolution object. For http_codes declared in the filter, the failure_type will default to `system_error`. - To override default failure_type use configured failure_type with ResponseAction.FAIL. + A configured failure_type overrides the default for the `FAIL`, `RETRY`, `RATE_LIMITED` and + `REFRESH_TOKEN_THEN_RETRY` actions. `IGNORE` and `RESET_PAGINATION` keep the default mapping. Attributes: action (Union[ResponseAction, str]): action to execute if a request matches @@ -95,7 +96,12 @@ def matches( error_message = self._create_error_message(response_or_exception) error_message = error_message or default_error_message - if self.failure_type and filter_action == ResponseAction.FAIL: + if self.failure_type and filter_action in { + ResponseAction.FAIL, + ResponseAction.RETRY, + ResponseAction.RATE_LIMITED, + ResponseAction.REFRESH_TOKEN_THEN_RETRY, + }: failure_type = self.failure_type elif default_mapped_error_resolution: failure_type = default_mapped_error_resolution.failure_type diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index c1d0eabd6..9ff8ca0db 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -316,7 +316,7 @@ def _send_with_retry( raise AirbyteTracedException( internal_message=f"Exhausted available request attempts. Exception: {e}", - message=f"Exhausted available request attempts. Please see logs for more details. Exception: {e}", + message="Available request retry attempts are exhausted.", failure_type=e.failure_type or FailureType.system_error, exception=e, stream_descriptor=StreamDescriptor(name=self._name), diff --git a/unit_tests/sources/declarative/requesters/error_handlers/test_http_response_filter.py b/unit_tests/sources/declarative/requesters/error_handlers/test_http_response_filter.py index 87e522d4a..02de1941d 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/test_http_response_filter.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/test_http_response_filter.py @@ -48,6 +48,21 @@ ), id="test_http_code_matches_ignore_action", ), + pytest.param( + ResponseAction.IGNORE, + FailureType.config_error, + {500}, + "", + "", + "", + {"status_code": 500}, + ErrorResolution( + response_action=ResponseAction.IGNORE, + failure_type=FailureType.transient_error, + error_message="HTTP Status Code: 500. Error: Internal server error.", + ), + id="test_http_code_matches_ignore_action_uses_default_failure_type", + ), pytest.param( ResponseAction.RETRY, None, @@ -178,10 +193,10 @@ {"status_code": 500}, ErrorResolution( response_action=ResponseAction.RETRY, - failure_type=FailureType.transient_error, + failure_type=FailureType.config_error, error_message="rate limits", ), - id="test_http_code_matches_failure_type_config_error_action_retry_uses_default_failure_type", + id="test_http_code_matches_failure_type_config_error_action_retry", ), pytest.param( ResponseAction.RATE_LIMITED, diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index 48d396cb6..ea415b790 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -11,6 +11,12 @@ from requests_cache import CachedRequest from airbyte_cdk.models import FailureType +from airbyte_cdk.sources.declarative.requesters.error_handlers import ( + DefaultErrorHandler as DeclarativeDefaultErrorHandler, +) +from airbyte_cdk.sources.declarative.requesters.error_handlers import ( + HttpResponseFilter, +) from airbyte_cdk.sources.streams.call_rate import CachedLimiterSession, LimiterSession from airbyte_cdk.sources.streams.http import HttpClient from airbyte_cdk.sources.streams.http.error_handlers import ( @@ -840,6 +846,46 @@ def backoff_time(self, response_or_exception, attempt_count): assert e.value.failure_type == expected_failure_type +@pytest.mark.usefixtures("mock_sleep") +def test_send_request_exhaustion_preserves_declared_failure_type_and_hides_retry_details( + requests_mock, +): + error_message = "The connector will retry automatically. Please see logs for more details." + error_handler = DeclarativeDefaultErrorHandler( + config={}, + parameters={}, + max_retries=1, + response_filters=[ + HttpResponseFilter( + action=ResponseAction.RETRY, + failure_type=FailureType.transient_error, + http_codes={200}, + error_message=error_message, + config={}, + parameters={}, + ) + ], + ) + http_client = HttpClient(name="test", logger=MagicMock(), error_handler=error_handler) + requests_mock.register_uri( + "GET", + "https://airbyte.io/", + status_code=200, + json={"code": 50000}, + headers={}, + ) + + with pytest.raises(AirbyteTracedException) as exception: + http_client.send_request(http_method="get", url="https://airbyte.io/", request_kwargs={}) + + assert exception.value.failure_type == FailureType.transient_error + assert exception.value.message == "Available request retry attempts are exhausted." + assert ( + exception.value.internal_message == "Exhausted available request attempts. Exception: " + f"{error_message}" + ) + + class MockOAuthAuthenticator: def __init__(self): self.access_token = "old_token"