[airbyte-cdk] Updates Low Code CDK ErrorHandlers and BackoffStrategies to align with Python CDK Interfaces

airbyte-cdk/python/airbyte_cdk/__init__.py

@@ -42,7 +42,6 @@
 from .sources.declarative.requesters import Requester, HttpRequester
 
 from .sources.declarative.requesters.error_handlers import BackoffStrategy
-from .sources.declarative.requesters.error_handlers.response_status import ResponseStatus
 from .sources.declarative.requesters.paginators import DefaultPaginator, PaginationStrategy
 from .sources.declarative.requesters.paginators.strategies import OffsetIncrement, CursorPaginationStrategy, PageIncrement, StopConditionPaginationStrategyDecorator
 

airbyte-cdk/python/airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1546,7 +1546,6 @@ definitions:
     type: object
     required:
       - type
-      - action
     properties:
       type:
         type: string

airbyte-cdk/python/airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -481,8 +481,8 @@ class Action(Enum):
 
 class HttpResponseFilter(BaseModel):
     type: Literal['HttpResponseFilter']
-    action: Action = Field(
-        ...,
+    action: Optional[Action] = Field(
+        None,
         description='Action to execute if a response matches the filter.',
         examples=['SUCCESS', 'FAIL', 'RETRY', 'IGNORE'],
         title='Action',
@@ -1086,7 +1086,7 @@ class Config:
     type: Literal['DeclarativeSource']
     check: CheckStream
     streams: List[DeclarativeStream]
-    version: str = Field(..., description='The version of the CDK used to build and test the source.')
+    version: str = Field(..., description='The version of the Airbyte CDK used to build and test the source.')
     schemas: Optional[Schemas] = None
     definitions: Optional[Dict[str, Any]] = None
     spec: Optional[Spec] = None

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

@@ -113,7 +113,6 @@
     WaitTimeFromHeaderBackoffStrategy,
     WaitUntilTimeFromHeaderBackoffStrategy,
 )
-from airbyte_cdk.sources.declarative.requesters.error_handlers.response_action import ResponseAction
 from airbyte_cdk.sources.declarative.requesters.paginators import DefaultPaginator, NoPagination, PaginatorTestReadDecorator
 from airbyte_cdk.sources.declarative.requesters.paginators.strategies import (
     CursorPaginationStrategy,
@@ -133,6 +132,7 @@
 from airbyte_cdk.sources.declarative.transformations import AddFields, RecordTransformation, RemoveFields
 from airbyte_cdk.sources.declarative.transformations.add_fields import AddedFieldDefinition
 from airbyte_cdk.sources.message import InMemoryMessageRepository, LogAppenderMessageRepositoryDecorator, MessageRepository
+from airbyte_cdk.sources.streams.http.error_handlers.response_models import ResponseAction
 from airbyte_cdk.sources.types import Config
 from airbyte_cdk.sources.utils.transform import TypeTransformer
 from isodate import parse_duration
@@ -333,7 +333,7 @@ def create_legacy_to_per_partition_state_migration(
         declarative_stream: DeclarativeStreamModel,
     ) -> LegacyToPerPartitionStateMigration:
         retriever = declarative_stream.retriever
-        partition_router = retriever.partition_router
+        partition_router = retriever.partition_router  # type: ignore # CustomRetriever would inherit from SimpleRetriever and therefore have partition_router
 
         if not isinstance(retriever, SimpleRetrieverModel):
             raise ValueError(
@@ -696,15 +696,7 @@ def create_default_error_handler(self, model: DefaultErrorHandlerModel, config:
             for response_filter_model in model.response_filters:
                 response_filters.append(self._create_component_from_model(model=response_filter_model, config=config))
         else:
-            response_filters.append(
-                HttpResponseFilter(
-                    ResponseAction.RETRY,
-                    http_codes=HttpResponseFilter.DEFAULT_RETRIABLE_ERRORS,
-                    config=config,
-                    parameters=model.parameters or {},
-                )
-            )
-            response_filters.append(HttpResponseFilter(ResponseAction.IGNORE, config=config, parameters=model.parameters or {}))
+            response_filters.append(HttpResponseFilter(config=config, parameters=model.parameters or {}))
 
         return DefaultErrorHandler(
             backoff_strategies=backoff_strategies,
@@ -800,7 +792,10 @@ def create_http_requester(self, model: HttpRequesterModel, config: Config, *, na
 
     @staticmethod
     def create_http_response_filter(model: HttpResponseFilterModel, config: Config, **kwargs: Any) -> HttpResponseFilter:
-        action = ResponseAction(model.action.value)
+        if model.action is not None:
+            action = ResponseAction(model.action.value)
+        else:
+            action = None
         http_codes = (
             set(model.http_codes) if model.http_codes else set()
         )  # JSON schema notation has no set data type. The schema enforces an array of unique elements

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

@@ -7,7 +7,7 @@
 
 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.streams.http.error_handlers import BackoffStrategy
 from airbyte_cdk.sources.types import Config
 
 
@@ -32,5 +32,5 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         else:
             self.backoff_time_in_seconds = InterpolatedString.create(self.backoff_time_in_seconds, parameters=parameters)
 
-    def backoff(self, response: requests.Response, attempt_count: int) -> Optional[float]:
+    def backoff_time(self, response_or_exception: Optional[Union[requests.Response, Exception]], attempt_count: int) -> Optional[float]:  # type: ignore # attempt_count maintained for compatibility with low code CDK
         return self.backoff_time_in_seconds.eval(self.config)  # type: ignore # backoff_time_in_seconds is always cast to an interpolated string

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

@@ -7,7 +7,7 @@
 
 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.streams.http.error_handlers import BackoffStrategy
 from airbyte_cdk.sources.types import Config
 
 
@@ -32,5 +32,7 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         else:
             self.factor = InterpolatedString.create(self.factor, parameters=parameters)
 
-    def backoff(self, response: requests.Response, attempt_count: int) -> Optional[float]:
+    def backoff_time(  # type: ignore # attempt_count maintained for compatibility with low code CDK
+        self, response_or_exception: Optional[Union[requests.Response, requests.RequestException]], attempt_count: int
+    ) -> Optional[float]:
         return self.factor.eval(self.config) * 2**attempt_count  # type: ignore # factor is always cast to an interpolated string

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

@@ -32,7 +32,7 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self.regex = InterpolatedString.create(self.regex, parameters=parameters) if self.regex else None
         self.header = InterpolatedString.create(self.header, parameters=parameters)
 
-    def backoff(self, response: requests.Response, attempt_count: int) -> Optional[float]:
+    def backoff_time(self, response: requests.Response, attempt_count: int) -> Optional[float]:  # type: ignore # attempt_count maintained for compatibility with low code CDK
         header = self.header.eval(config=self.config)  # type: ignore  # header is always cast to an interpolated stream
         if self.regex:
             evaled_regex = self.regex.eval(self.config)  # type: ignore # header is always cast to an interpolated string

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

@@ -39,15 +39,17 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         if not isinstance(self.min_wait, InterpolatedString):
             self.min_wait = InterpolatedString.create(str(self.min_wait), parameters=parameters)
 
-    def backoff(self, response: requests.Response, attempt_count: int) -> Optional[float]:
+    def backoff_time(self, response_or_exception: Optional[Union[requests.Response, requests.RequestException]], attempt_count: int) -> Optional[float]:  # type: ignore # attempt_count maintained for compatibility with low code CDK
         now = time.time()
         header = self.header.eval(self.config)  # type: ignore # header is always cast to an interpolated string
         if self.regex:
             evaled_regex = self.regex.eval(self.config)  # type: ignore # header is always cast to an interpolated string
             regex = re.compile(evaled_regex)
         else:
             regex = None
-        wait_until = get_numeric_value_from_header(response, header, regex)
+        wait_until = None
+        if isinstance(response_or_exception, requests.Response):
+            wait_until = get_numeric_value_from_header(response_or_exception, header, regex)
         min_wait = self.min_wait.eval(self.config)  # type: ignore # header is always cast to an interpolated string
         if wait_until is None or not wait_until:
             return float(min_wait) if min_wait else None

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

@@ -2,24 +2,13 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-from abc import abstractmethod
-from dataclasses import dataclass
-from typing import Optional
+from abc import ABC
 
-import requests
+from airbyte_cdk.sources.streams.http.error_handlers import BackoffStrategy as HttpBackoffStrategy
 
 
-@dataclass
-class BackoffStrategy:
+class BackoffStrategy(HttpBackoffStrategy, ABC):
     """
     Backoff strategy defining how long to wait before retrying a request that resulted in an error.
+    References Python CDK BackoffStrategy
     """
-
-    @abstractmethod
-    def backoff(self, response: requests.Response, attempt_count: int) -> Optional[float]:
-        """
-        Return time to wait before retrying the request.
-        :param response: response received for the request to retry
-        :param attempt_count: number of attempts to submit the request
-        :return: time to wait in seconds
-        """

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

@@ -3,13 +3,11 @@
 #
 
 from dataclasses import InitVar, dataclass
-from typing import Any, List, Mapping, Union
+from typing import Any, List, Mapping, Optional, Union
 
-import airbyte_cdk.sources.declarative.requesters.error_handlers.response_status as response_status
 import requests
-from airbyte_cdk.sources.declarative.requesters.error_handlers.error_handler import ErrorHandler
-from airbyte_cdk.sources.declarative.requesters.error_handlers.response_action import ResponseAction
-from airbyte_cdk.sources.declarative.requesters.error_handlers.response_status import ResponseStatus
+from airbyte_cdk.sources.streams.http.error_handlers import ErrorHandler
+from airbyte_cdk.sources.streams.http.error_handlers.response_models import DEFAULT_ERROR_RESOLUTION, ErrorResolution, ResponseAction
 
 
 @dataclass
@@ -45,19 +43,28 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
             raise ValueError("CompositeErrorHandler expects at least 1 underlying error handler")
 
     @property
-    def max_retries(self) -> Union[int, None]:
-        return self.error_handlers[0].max_retries
+    def max_retries(self) -> Optional[int]:
+        return self.error_handlers[0].max_retries  # type: ignore # property not defined in ErrorHandler
 
     @property
     def max_time(self) -> Union[int, None]:
-        return max([error_handler.max_time or 0 for error_handler in self.error_handlers])
-
-    def interpret_response(self, response: requests.Response) -> ResponseStatus:
-        should_retry = ResponseStatus(ResponseAction.FAIL)
-        for retrier in self.error_handlers:
-            should_retry = retrier.interpret_response(response)
-            if should_retry.action == ResponseAction.SUCCESS:
-                return response_status.SUCCESS
-            if should_retry == response_status.IGNORE or should_retry.action == ResponseAction.RETRY:
-                return should_retry
-        return should_retry
+        return max([error_handler.max_time or 0 for error_handler in self.error_handlers])  # type: ignore # property not defined in ErrorHandler
+
+    def interpret_response(self, response_or_exception: Optional[Union[requests.Response, Exception]]) -> ErrorResolution:
+        matched_error_resolution = None
+        for error_handler in self.error_handlers:
+            matched_error_resolution = error_handler.interpret_response(response_or_exception)
+
+            if matched_error_resolution.response_action == ResponseAction.SUCCESS:
+                return matched_error_resolution
+
+            if (
+                matched_error_resolution.response_action == ResponseAction.RETRY
+                or matched_error_resolution.response_action == ResponseAction.IGNORE
+            ):
+                return matched_error_resolution
+
+        if matched_error_resolution is not None:
+            return matched_error_resolution
+
+        return DEFAULT_ERROR_RESOLUTION

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

@@ -5,16 +5,13 @@
 from dataclasses import InitVar, dataclass, field
 from typing import Any, List, Mapping, MutableMapping, Optional, Union
 
-import airbyte_cdk.sources.declarative.requesters.error_handlers.response_status as response_status
 import requests
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.exponential_backoff_strategy import (
     ExponentialBackoffStrategy,
 )
-from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategy import BackoffStrategy
-from airbyte_cdk.sources.declarative.requesters.error_handlers.error_handler import ErrorHandler
 from airbyte_cdk.sources.declarative.requesters.error_handlers.http_response_filter import HttpResponseFilter
-from airbyte_cdk.sources.declarative.requesters.error_handlers.response_action import ResponseAction
-from airbyte_cdk.sources.declarative.requesters.error_handlers.response_status import ResponseStatus
+from airbyte_cdk.sources.streams.http.error_handlers import BackoffStrategy, ErrorHandler
+from airbyte_cdk.sources.streams.http.error_handlers.response_models import DEFAULT_ERROR_RESOLUTION, SUCCESS_RESOLUTION, ErrorResolution
 from airbyte_cdk.sources.types import Config
 
 
@@ -101,57 +98,46 @@ class DefaultErrorHandler(ErrorHandler):
     backoff_strategies: Optional[List[BackoffStrategy]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
-        self.response_filters = self.response_filters or []
 
         if not self.response_filters:
-            self.response_filters.append(
-                HttpResponseFilter(
-                    ResponseAction.RETRY, http_codes=HttpResponseFilter.DEFAULT_RETRIABLE_ERRORS, config=self.config, parameters={}
-                )
-            )
-            self.response_filters.append(HttpResponseFilter(ResponseAction.IGNORE, config={}, parameters={}))
+            self.response_filters = [HttpResponseFilter(config=self.config, parameters={})]
 
         if not self.backoff_strategies:
             self.backoff_strategies = [DefaultErrorHandler.DEFAULT_BACKOFF_STRATEGY(parameters=parameters, config=self.config)]
 
         self._last_request_to_attempt_count: MutableMapping[requests.PreparedRequest, int] = {}
 
-    @property  # type: ignore # overwrite the property to handle the case where max_retries is not provided in the constructor
-    def max_retries(self) -> Union[int, None]:
-        return self._max_retries
-
-    @max_retries.setter
-    def max_retries(self, value: int) -> None:
-        # Covers the case where max_retries is not provided in the constructor, which causes the property object
-        # to be set which we need to avoid doing
-        if not isinstance(value, property):
-            self._max_retries = value
-
-    def interpret_response(self, response: requests.Response) -> ResponseStatus:
-        request = response.request
-
-        if request not in self._last_request_to_attempt_count:
-            self._last_request_to_attempt_count = {request: 1}
-        else:
-            self._last_request_to_attempt_count[request] += 1
-        if self.response_filters:
-            for response_filter in self.response_filters:
-                matched_status = response_filter.matches(
-                    response=response, backoff_time=self._backoff_time(response, self._last_request_to_attempt_count[request])
-                )
-                if matched_status is not None:
-                    return matched_status
-
-        if response.ok:
-            return response_status.SUCCESS
-        # Fail if the response matches no filters
-        return response_status.FAIL
-
-    def _backoff_time(self, response: requests.Response, attempt_count: int) -> Optional[float]:
+    def interpret_response(self, response_or_exception: Optional[Union[requests.Response, Exception]]) -> ErrorResolution:
+
+        if isinstance(response_or_exception, requests.Response):
+
+            request = response_or_exception.request
+
+            if request not in self._last_request_to_attempt_count:
+                self._last_request_to_attempt_count = {request: 1}
+            else:
+                self._last_request_to_attempt_count[request] += 1
+            if self.response_filters:
+
+                for response_filter in self.response_filters:
+                    matched_error_resolution = response_filter.matches(response_or_exception=response_or_exception)
+
+                    if matched_error_resolution is not None:
+                        return matched_error_resolution
+
+            if response_or_exception.ok:
+                return SUCCESS_RESOLUTION
+
+        # Return default error resolution (retry)
+        return DEFAULT_ERROR_RESOLUTION
+
+    def backoff_time(
+        self, response_or_exception: Optional[Union[requests.Response, requests.RequestException]], attempt_count: int = 0
+    ) -> Optional[float]:
         backoff = None
         if self.backoff_strategies:
             for backoff_strategies in self.backoff_strategies:
-                backoff = backoff_strategies.backoff(response, attempt_count)
+                backoff = backoff_strategies.backoff_time(response_or_exception, attempt_count)  # type: ignore # attempt_count maintained for compatibility with low code CDK
                 if backoff:
                     return backoff
         return backoff

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

@@ -2,42 +2,15 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-from abc import abstractmethod
+from abc import ABC
 from dataclasses import dataclass
-from typing import Union
 
-import requests
-from airbyte_cdk.sources.declarative.requesters.error_handlers.response_status import ResponseStatus
+from airbyte_cdk.sources.streams.http.error_handlers import ErrorHandler as HttpErrorHandler
 
 
 @dataclass
-class ErrorHandler:
+class ErrorHandler(HttpErrorHandler, ABC):
     """
     Defines whether a request was successful and how to handle a failure.
+    References Python CDK ErrorHandler
     """
-
-    @property
-    @abstractmethod
-    def max_retries(self) -> Union[int, None]:
-        """
-        Specifies maximum amount of retries for backoff policy. Return None for no limit.
-        """
-        pass
-
-    @property
-    @abstractmethod
-    def max_time(self) -> Union[int, None]:
-        """
-        Specifies maximum total waiting time (in seconds) for backoff policy. Return None for no limit.
-        """
-        pass
-
-    @abstractmethod
-    def interpret_response(self, response: requests.Response) -> ResponseStatus:
-        """
-        Evaluate response status describing whether a failing request should be retried or ignored.
-
-        :param response: response to evaluate
-        :return: response status
-        """
-        pass