From 9aef805f92331bec7078e075e30625ac37912c62 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 6 Oct 2026 15:07:02 +0000 Subject: [PATCH] feat(low-code): add response, headers and attempt_count to ConstantBackoffStrategy interpolation Co-Authored-By: bot_apk --- .../declarative_component_schema.yaml | 10 +- .../models/declarative_component_schema.py | 11 +- .../constant_backoff_strategy.py | 61 ++++++- .../test_constant_backoff.py | 161 +++++++++++++++++- .../test_default_error_handler.py | 55 ++++++ 5 files changed, 289 insertions(+), 9 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index dfabccd6b0..414ed654fe 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -698,7 +698,7 @@ definitions: additionalProperties: true ConstantBackoffStrategy: title: Constant Backoff - description: Backoff strategy with a constant backoff interval. + description: Backoff strategy with a constant backoff interval. The backoff time can be computed from the response via interpolation. If the rendered value is empty, None, 0, or non-numeric, this strategy is skipped (no jitter is applied) so the next entry in backoff_strategies is used. type: object required: - type @@ -709,7 +709,7 @@ definitions: enum: [ConstantBackoffStrategy] backoff_time_in_seconds: title: Backoff Time - description: Backoff time in seconds. + description: Backoff time in seconds. Can be computed from the response via interpolation; if the rendered value is empty, None, 0, or non-numeric, this strategy is skipped and the next backoff strategy is used. anyOf: - type: number title: Number of seconds @@ -718,10 +718,16 @@ definitions: title: Interpolated Value interpolation_context: - config + - response + - headers + - attempt_count examples: - 30 - 30.5 - "{{ config['backoff_time'] }}" + - "{{ 900 if 'load quota exceeded' in ((response.get('error') or {}).get('message') or '') | lower else 0 }}" + - "{{ (1 / (headers.get('x-amzn-RateLimit-Limit') | float)) if (headers.get('x-amzn-RateLimit-Limit') | float) > 0 else 60 }}" + - "{{ (response.get('message', '') | regex_search('(?i)Retry after\\s+(\\d+)\\s+seconds')) or min(2 ** attempt_count, 120) if response is mapping else min(2 ** attempt_count, 120) }}" jitter_range_in_seconds: title: Jitter Range description: Optional additive jitter range in seconds. When set, the backoff time is uniformly distributed between backoff_time_in_seconds and backoff_time_in_seconds + (jitter_range_in_seconds * 2), so jitter only increases the base backoff. diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 5581be3757..e0b1d864ec 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -136,8 +136,15 @@ class ConstantBackoffStrategy(BaseModel): type: Literal["ConstantBackoffStrategy"] backoff_time_in_seconds: Union[float, str] = Field( ..., - description="Backoff time in seconds.", - examples=[30, 30.5, "{{ config['backoff_time'] }}"], + description="Backoff time in seconds. Can be computed from the response via interpolation; if the rendered value is empty, None, 0, or non-numeric, this strategy is skipped and the next backoff strategy is used.", + examples=[ + 30, + 30.5, + "{{ config['backoff_time'] }}", + "{{ 900 if 'load quota exceeded' in ((response.get('error') or {}).get('message') or '') | lower else 0 }}", + "{{ (1 / (headers.get('x-amzn-RateLimit-Limit') | float)) if (headers.get('x-amzn-RateLimit-Limit') | float) > 0 else 60 }}", + "{{ (response.get('message', '') | regex_search('(?i)Retry after\\s+(\\d+)\\s+seconds')) or min(2 ** attempt_count, 120) if response is mapping else min(2 ** attempt_count, 120) }}", + ], title="Backoff Time", ) jitter_range_in_seconds: Optional[float] = Field( diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py index 931183b2ef..fea6b0fa85 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py @@ -2,21 +2,40 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # +import logging import random from dataclasses import InitVar, dataclass from typing import Any, Mapping, Optional, Union, cast import requests +from jinja2.exceptions import UndefinedError from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString from airbyte_cdk.sources.streams.http.error_handlers import BackoffStrategy from airbyte_cdk.sources.types import Config +logger = logging.getLogger("airbyte") + @dataclass class ConstantBackoffStrategy(BackoffStrategy): """ - Backoff strategy with a constant backoff interval + Backoff strategy with a constant backoff interval. + + The ``backoff_time_in_seconds`` field is evaluated on every call with the + following interpolation context, in addition to ``config``: + + * ``response``: the parsed JSON body of the response (``{}`` when the body is + not JSON or when the call was made with a request exception or ``None``). + * ``headers``: the response headers (``{}`` on a request exception). + * ``attempt_count``: the number of the current retry attempt. + + If the expression cannot be evaluated against the response (for example an + attribute access on a non-object JSON body), the rendered value is falsy + (``None``, ``""``, ``0``), evaluates to ``0`` seconds, or cannot be + converted to a number, this strategy returns ``None`` (no jitter is + applied) so that the next entry in ``backoff_strategies`` or the default + backoff is used. Attributes: backoff_time_in_seconds (float): time to backoff before retrying a retryable request. @@ -44,10 +63,44 @@ def backoff_time( response_or_exception: Optional[Union[requests.Response, requests.RequestException]], attempt_count: int, ) -> Optional[float]: - backoff_time = float( - cast(InterpolatedString, self.backoff_time_in_seconds).eval(self.config) - ) + response: Any + headers: Mapping[str, Any] + if isinstance(response_or_exception, requests.Response): + response = self._safe_response_json(response_or_exception) + headers = response_or_exception.headers + else: + response = {} + headers = {} + + try: + rendered = cast(InterpolatedString, self.backoff_time_in_seconds).eval( + self.config, + response=response, + headers=headers, + attempt_count=attempt_count, + ) + except UndefinedError: + return None + if not rendered: + return None + try: + backoff_time = float(rendered) + except (TypeError, ValueError): + logger.warning( + f"ConstantBackoffStrategy backoff_time_in_seconds rendered to a non-numeric value {rendered!r}; " + "skipping this strategy so the next backoff strategy or the default backoff is used." + ) + return None + if backoff_time == 0: + return None if self.jitter_range_in_seconds is None: return backoff_time return random.uniform(backoff_time, backoff_time + (self.jitter_range_in_seconds * 2)) + + @staticmethod + def _safe_response_json(response: requests.Response) -> Any: + try: + return response.json() + except requests.exceptions.JSONDecodeError: + return {} diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py index 7fae93fa15..1f40b2276f 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py @@ -2,10 +2,18 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # +import json from unittest.mock import MagicMock import pytest +import requests +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + ConstantBackoffStrategy as ConstantBackoffStrategyModel, +) +from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import ( + ModelToComponentFactory, +) from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy import ( ConstantBackoffStrategy, ) @@ -14,6 +22,34 @@ PARAMETERS_BACKOFF_TIME = 20 CONFIG_BACKOFF_TIME = 30 +GSC_BACKOFF_EXPRESSION = "{{ 900 if 'load quota exceeded' in ((response.get('error') or {}).get('message') or '') | lower else 0 }}" +AMAZON_BACKOFF_EXPRESSION = "{{ (1 / (headers.get('x-amzn-RateLimit-Limit') | float)) if (headers.get('x-amzn-RateLimit-Limit') | float) > 0 else 60 }}" +PINTEREST_BACKOFF_EXPRESSION = "{{ (response.get('message', '') | regex_search('(?i)Retry after\\s+(\\d+)\\s+seconds')) or min(2 ** attempt_count, 120) if response is mapping else min(2 ** attempt_count, 120) }}" +LINKEDIN_BACKOFF_EXPRESSION = "{{ 330 if 'data request limit has been exceeded' in (response.get('message') or '') | lower and '45 million metric values' in (response.get('message') or '') | lower else 0 }}" + + +def _make_response(status_code=429, json_body=None, raw_body=None, headers=None): + response = requests.Response() + response.status_code = status_code + response.url = "https://airbyte.io" + response.headers.update(headers or {}) + if raw_body is not None: + response._content = raw_body + elif json_body is not None: + response._content = json.dumps(json_body).encode("utf-8") + else: + response._content = b"{}" + return response + + +def _strategy(backoff_time_in_seconds, jitter_range=None, parameters=None, config=None): + return ConstantBackoffStrategy( + parameters=parameters or {}, + backoff_time_in_seconds=backoff_time_in_seconds, + jitter_range_in_seconds=jitter_range, + config=config or {}, + ) + @pytest.mark.parametrize( "test_name, attempt_count, backofftime, jitter_range, expected_backoff_time", @@ -60,7 +96,6 @@ def test_constant_backoff( [ pytest.param(60, 15, 60, 90, id="base_backoff_floor"), pytest.param(10, 30, 10, 70, id="large_jitter"), - pytest.param(0, 5, 0, 10, id="zero_base"), ], ) def test_constant_backoff_with_jitter_bounds( @@ -80,3 +115,127 @@ def test_constant_backoff_with_jitter_bounds( assert all(expected_lower_bound <= backoff <= expected_upper_bound for backoff in backoff_times) assert len(set(backoff_times)) > 1 + + +@pytest.mark.parametrize( + "json_body, expected_backoff_time", + [ + pytest.param( + {"error": {"message": "Search Analytics load quota exceeded."}}, + 900, + id="load_quota_exceeded", + ), + pytest.param( + {"error": {"message": "Search Analytics QPS quota exceeded."}}, + None, + id="qps_quota_exceeded", + ), + ], +) +def test_constant_backoff_response_body(json_body, expected_backoff_time): + backoff_strategy = _strategy(GSC_BACKOFF_EXPRESSION) + response = _make_response(json_body=json_body) + assert backoff_strategy.backoff_time(response, attempt_count=1) == expected_backoff_time + + +@pytest.mark.parametrize( + "headers, expected_backoff_time", + [ + pytest.param({"x-amzn-RateLimit-Limit": "0.5"}, 2.0, id="low_rate_limit"), + pytest.param({"x-amzn-RateLimit-Limit": "2.0"}, 0.5, id="higher_rate_limit"), + pytest.param({}, 60, id="missing_rate_limit_header"), + pytest.param({"x-amzn-RateLimit-Limit": "abc"}, 60, id="garbage_rate_limit_header"), + ], +) +def test_constant_backoff_response_headers(headers, expected_backoff_time): + backoff_strategy = _strategy(AMAZON_BACKOFF_EXPRESSION) + response = _make_response(headers=headers) + assert backoff_strategy.backoff_time(response, attempt_count=1) == expected_backoff_time + + +@pytest.mark.parametrize( + "json_body, raw_body, attempt_count, expected_backoff_time", + [ + pytest.param( + {"message": "Rate limited. Retry after 30 seconds"}, + None, + 1, + 30, + id="retry_after_in_body", + ), + pytest.param({"message": "rate limited"}, None, 2, 4, id="exponential_fallback"), + pytest.param({"message": "rate limited"}, None, 10, 120, id="exponential_fallback_capped"), + pytest.param([1, 2, 3], None, 2, 4, id="non_dict_json_body"), + pytest.param(None, b"not json at all", 3, 8, id="non_json_body"), + ], +) +def test_constant_backoff_response_body_and_attempt_count( + json_body, raw_body, attempt_count, expected_backoff_time +): + backoff_strategy = _strategy(PINTEREST_BACKOFF_EXPRESSION) + response = _make_response(json_body=json_body, raw_body=raw_body) + assert ( + backoff_strategy.backoff_time(response, attempt_count=attempt_count) + == expected_backoff_time + ) + + +def test_constant_backoff_unguarded_get_on_list_body_falls_through(): + backoff_strategy = _strategy("{{ response.get('message') }}") + response = _make_response(json_body=[1, 2, 3]) + assert backoff_strategy.backoff_time(response, attempt_count=1) is None + + +@pytest.mark.parametrize("jitter_range", [None, 15]) +@pytest.mark.parametrize( + "backofftime", + [ + pytest.param("{{ '' }}", id="empty_string_render"), + pytest.param("{{ None }}", id="none_render"), + pytest.param("{{ 0 }}", id="zero_render"), + pytest.param(0, id="plain_zero"), + ], +) +def test_constant_backoff_falsy_render_falls_through(backofftime, jitter_range): + backoff_strategy = _strategy(backofftime, jitter_range=jitter_range) + response = _make_response() + assert backoff_strategy.backoff_time(response, attempt_count=1) is None + + +def test_constant_backoff_non_numeric_render_falls_through(caplog): + backoff_strategy = _strategy("{{ 'abc' }}") + response = _make_response() + assert backoff_strategy.backoff_time(response, attempt_count=1) is None + + +@pytest.mark.parametrize( + "response_or_exception", + [ + pytest.param(requests.exceptions.ConnectionError(), id="request_exception"), + pytest.param(None, id="no_response"), + ], +) +def test_constant_backoff_on_exception_or_none(response_or_exception): + backoff_strategy = _strategy("{{ headers.get('x') or response.get('y') or 7 }}") + assert backoff_strategy.backoff_time(response_or_exception, attempt_count=1) == 7 + + +def test_constant_backoff_from_factory(): + manifest = { + "type": "ConstantBackoffStrategy", + "backoff_time_in_seconds": LINKEDIN_BACKOFF_EXPRESSION, + } + backoff_strategy = ModelToComponentFactory().create_component( + ConstantBackoffStrategyModel, manifest, config={} + ) + assert isinstance(backoff_strategy, ConstantBackoffStrategy) + + throttled_response = _make_response( + json_body={ + "message": "Data request limit has been exceeded. Limit of 45 million metric values per day." + } + ) + assert backoff_strategy.backoff_time(throttled_response, attempt_count=1) == 330 + + other_response = _make_response(json_body={"message": "some other error"}) + assert backoff_strategy.backoff_time(other_response, attempt_count=1) is None diff --git a/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py b/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py index 526e5760fc..4df25f0980 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py @@ -2,6 +2,7 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # +import json from unittest.mock import MagicMock import pytest @@ -242,6 +243,60 @@ def test_default_error_handler_with_constant_backoff_strategy(): ) +GSC_BACKOFF_EXPRESSION = "{{ 900 if 'load quota exceeded' in ((response.get('error') or {}).get('message') or '') | lower else 0 }}" + + +def _real_response(status_code, json_body=None, headers=None): + response = requests.Response() + response.status_code = status_code + response.url = "https://airbyte.io" + response.headers.update(headers or {}) + response._content = json.dumps(json_body or {}).encode("utf-8") + return response + + +def test_default_error_handler_falls_through_to_next_backoff_strategy(): + error_handler = DefaultErrorHandler( + config={}, + parameters={}, + backoff_strategies=[ + ConstantBackoffStrategy(GSC_BACKOFF_EXPRESSION, config={}, parameters={}), + ConstantBackoffStrategy(60, jitter_range_in_seconds=15, config={}, parameters={}), + ], + ) + + load_quota_response = _real_response( + 429, json_body={"error": {"message": "Search Analytics load quota exceeded."}} + ) + assert ( + error_handler.backoff_time(response_or_exception=load_quota_response, attempt_count=0) + == 900 + ) + + qps_response = _real_response( + 429, json_body={"error": {"message": "Search Analytics QPS quota exceeded."}} + ) + for _ in range(50): + backoff = error_handler.backoff_time(response_or_exception=qps_response, attempt_count=0) + assert 60 <= backoff <= 90 + + +def test_default_error_handler_falsy_render_with_jitter_falls_through(): + error_handler = DefaultErrorHandler( + config={}, + parameters={}, + backoff_strategies=[ + ConstantBackoffStrategy( + "{{ 0 }}", jitter_range_in_seconds=15, config={}, parameters={} + ), + ExponentialBackoffStrategy(factor=5, config={}, parameters={}), + ], + ) + + response = _real_response(429) + assert error_handler.backoff_time(response_or_exception=response, attempt_count=0) == 5 + + @pytest.mark.parametrize( "attempt_count", [