[low-code]: Evaluate backoff strategies at runtime

airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/factory.py

@@ -229,14 +229,17 @@ def _create_subcomponent(self, key, definition, kwargs, config, parent_class, in
                 options = kwargs.get(OPTIONS_STR, {})
                 try:
                     # enums can't accept options
-                    if issubclass(expected_type, enum.Enum):
+                    if issubclass(expected_type, enum.Enum) or self.isBuiltinTypes(definition):
                         return expected_type(definition)
                     else:
                         return expected_type(definition, options=options)
                 except Exception as e:
                     raise Exception(f"failed to instantiate type {expected_type}. {e}")
         return definition
 
+    def isBuiltinTypes(self, obj):
+        return isinstance(obj, (int, float, bool))
+
     @staticmethod
     def is_object_definition_with_class_name(definition):
         return isinstance(definition, dict) and "class_name" in definition

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py

@@ -2,11 +2,13 @@
 # Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 #
 
-from dataclasses import dataclass
-from typing import Optional
+from dataclasses import InitVar, dataclass
+from typing import Any, Mapping, Optional, Union
 
 import requests
+from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategy import BackoffStrategy
+from airbyte_cdk.sources.declarative.types import Config
 from dataclasses_jsonschema import JsonSchemaMixin
 
 
@@ -19,7 +21,14 @@ class ConstantBackoffStrategy(BackoffStrategy, JsonSchemaMixin):
         backoff_time_in_seconds (float): time to backoff before retrying a retryable request.
     """
 
-    backoff_time_in_seconds: float
+    backoff_time_in_seconds: Union[float, InterpolatedString, str]
+    options: InitVar[Mapping[str, Any]]
+    config: Config
+
+    def __post_init__(self, options: Mapping[str, Any]):
+        if not isinstance(self.backoff_time_in_seconds, InterpolatedString):
+            self.backoff_time_in_seconds = str(self.backoff_time_in_seconds)
+        self.backoff_time_in_seconds = InterpolatedString.create(self.backoff_time_in_seconds, options=options)
 
     def backoff(self, response: requests.Response, attempt_count: int) -> Optional[float]:
-        return self.backoff_time_in_seconds
+        return self.backoff_time_in_seconds.eval(self.config)

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py

@@ -3,8 +3,8 @@
 #
 
 import re
-from dataclasses import dataclass
-from typing import Optional
+from dataclasses import InitVar, dataclass
+from typing import Any, Mapping, Optional
 
 import requests
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.header_helper import get_numeric_value_from_header
@@ -23,9 +23,10 @@ class WaitTimeFromHeaderBackoffStrategy(BackoffStrategy, JsonSchemaMixin):
     """
 
     header: str
+    options: InitVar[Mapping[str, Any]]
     regex: Optional[str] = None
 
-    def __post_init__(self):
+    def __post_init__(self, options: Mapping[str, Any]):
         self.regex = re.compile(self.regex) if self.regex else None
 
     def backoff(self, response: requests.Response, attempt_count: int) -> Optional[float]:

airbyte-cdk/python/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py

@@ -8,17 +8,23 @@
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy import ConstantBackoffStrategy
 
 BACKOFF_TIME = 10
+OPTIONS_BACKOFF_TIME = 20
+CONFIG_BACKOFF_TIME = 30
 
 
 @pytest.mark.parametrize(
-    "test_name, attempt_count, expected_backoff_time",
+    "test_name, attempt_count, backofftime, expected_backoff_time",
     [
-        ("test_exponential_backoff", 1, BACKOFF_TIME),
-        ("test_exponential_backoff", 2, BACKOFF_TIME),
+        ("test_constant_backoff_first_attempt", 1, BACKOFF_TIME, BACKOFF_TIME),
+        ("test_constant_backoff_second_attempt", 2, BACKOFF_TIME, BACKOFF_TIME),
+        ("test_constant_backoff_from_options", 1, "{{ options['backoff'] }}", OPTIONS_BACKOFF_TIME),
+        ("test_constant_backoff_from_config", 1, "{{ config['backoff'] }}", CONFIG_BACKOFF_TIME),
     ],
 )
-def test_exponential_backoff(test_name, attempt_count, expected_backoff_time):
+def test_constant_backoff(test_name, attempt_count, backofftime, expected_backoff_time):
     response_mock = MagicMock()
-    backoff_strategy = ConstantBackoffStrategy(backoff_time_in_seconds=BACKOFF_TIME)
+    backoff_strategy = ConstantBackoffStrategy(
+        options={"backoff": OPTIONS_BACKOFF_TIME}, backoff_time_in_seconds=backofftime, config={"backoff": CONFIG_BACKOFF_TIME}
+    )
     backoff = backoff_strategy.backoff(response_mock, attempt_count)
     assert backoff == expected_backoff_time

airbyte-cdk/python/unit_tests/sources/declarative/requesters/error_handlers/test_default_error_handler.py

@@ -30,7 +30,7 @@
             None,
             {},
             ResponseStatus.retry(SOME_BACKOFF_TIME),
-            [ConstantBackoffStrategy(SOME_BACKOFF_TIME)],
+            [ConstantBackoffStrategy(options={}, backoff_time_in_seconds=SOME_BACKOFF_TIME, config={})],
         ),
         ("test_exponential_backoff", HTTPStatus.BAD_GATEWAY, None, None, {}, ResponseStatus.retry(10), None),
         (
@@ -50,7 +50,10 @@
             None,
             {},
             ResponseStatus.retry(10),
-            [DefaultErrorHandler.DEFAULT_BACKOFF_STRATEGY(), ConstantBackoffStrategy(SOME_BACKOFF_TIME)],
+            [
+                DefaultErrorHandler.DEFAULT_BACKOFF_STRATEGY(),
+                ConstantBackoffStrategy(options={}, backoff_time_in_seconds=SOME_BACKOFF_TIME, config={}),
+            ],
         ),
         ("test_200", HTTPStatus.OK, None, None, {}, response_status.SUCCESS, None),
         ("test_3XX", HTTPStatus.PERMANENT_REDIRECT, None, None, {}, response_status.SUCCESS, None),

airbyte-cdk/python/unit_tests/sources/declarative/test_factory.py

@@ -414,31 +414,31 @@ def test_create_record_selector(test_name, record_selector, expected_runtime_sel
         (
             "test_option_in_selector",
             """
-      extractor:
-        type: DpathExtractor
-        field_pointer: ["{{ options['name'] }}"]
-      selector:
-        class_name: airbyte_cdk.sources.declarative.extractors.record_selector.RecordSelector
-        $options:
-          name: "selector"
-        extractor: "*ref(extractor)"
-    """,
+          extractor:
+            type: DpathExtractor
+            field_pointer: ["{{ options['name'] }}"]
+          selector:
+            class_name: airbyte_cdk.sources.declarative.extractors.record_selector.RecordSelector
+            $options:
+              name: "selector"
+            extractor: "*ref(extractor)"
+        """,
             "selector",
         ),
         (
             "test_option_in_extractor",
             """
-      extractor:
-        type: DpathExtractor
-        $options:
-          name: "extractor"
-        field_pointer: ["{{ options['name'] }}"]
-      selector:
-        class_name: airbyte_cdk.sources.declarative.extractors.record_selector.RecordSelector
-        $options:
-          name: "selector"
-        extractor: "*ref(extractor)"
-    """,
+          extractor:
+            type: DpathExtractor
+            $options:
+              name: "extractor"
+            field_pointer: ["{{ options['name'] }}"]
+          selector:
+            class_name: airbyte_cdk.sources.declarative.extractors.record_selector.RecordSelector
+            $options:
+              name: "selector"
+            extractor: "*ref(extractor)"
+        """,
             "extractor",
         ),
     ],
@@ -467,6 +467,10 @@ def test_create_requester():
         a_parameter: "something_here"
       request_headers:
         header: header_value
+    error_handler:
+      backoff_strategies:
+        - type: "ConstantBackoffStrategy"
+          backoff_time_in_seconds: 5
     """
     config = parser.parse(content)