airbyte_cdk.sources.declarative.requesters.error_handlers

12class BackoffStrategy(ABC):
13    @abstractmethod
14    def backoff_time(
15        self,
16        response_or_exception: Optional[Union[requests.Response, requests.RequestException]],
17        attempt_count: int,
18    ) -> Optional[float]:
19        """
20        Override this method to dynamically determine backoff time e.g: by reading the X-Retry-After header.
21
22        This method is called only if should_backoff() returns True for the input request.
23
24        :param response_or_exception: The response or exception that caused the backoff.
25        :param attempt_count: The number of attempts already performed for this request.
26        :return how long to backoff in seconds. The return value may be a floating point number for subsecond precision. Returning None defers backoff
27        to the default backoff behavior (e.g using an exponential algorithm).
28        """
29        pass

 20@dataclass
 21class CompositeErrorHandler(ErrorHandler):
 22    """
 23    Error handler that sequentially iterates over a list of `ErrorHandler`s
 24
 25    Sample config chaining 2 different retriers:
 26        error_handler:
 27          type: "CompositeErrorHandler"
 28          error_handlers:
 29            - response_filters:
 30                - predicate: "{{ 'codase' in response }}"
 31                  action: RETRY
 32              backoff_strategies:
 33                - type: "ConstantBackoff"
 34                  backoff_time_in_seconds: 5
 35            - response_filters:
 36                - http_codes: [ 403 ]
 37                  action: RETRY
 38              backoff_strategies:
 39                - type: "ConstantBackoff"
 40                  backoff_time_in_seconds: 10
 41    Attributes:
 42        error_handlers (List[ErrorHandler]): list of error handlers
 43    """
 44
 45    error_handlers: List[ErrorHandler]
 46    parameters: InitVar[Mapping[str, Any]]
 47
 48    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 49        if not self.error_handlers:
 50            raise ValueError("CompositeErrorHandler expects at least 1 underlying error handler")
 51
 52    @property
 53    def max_retries(self) -> Optional[int]:
 54        return self.error_handlers[0].max_retries
 55
 56    @property
 57    def max_time(self) -> Optional[int]:
 58        return max([error_handler.max_time or 0 for error_handler in self.error_handlers])
 59
 60    def interpret_response(
 61        self, response_or_exception: Optional[Union[requests.Response, Exception]]
 62    ) -> ErrorResolution:
 63        matched_error_resolution = None
 64        for error_handler in self.error_handlers:
 65            matched_error_resolution = error_handler.interpret_response(response_or_exception)
 66
 67            if not isinstance(matched_error_resolution, ErrorResolution):
 68                continue
 69
 70            if matched_error_resolution.response_action in [
 71                ResponseAction.SUCCESS,
 72                ResponseAction.RETRY,
 73                ResponseAction.IGNORE,
 74                ResponseAction.RESET_PAGINATION,
 75            ]:
 76                return matched_error_resolution
 77
 78        if matched_error_resolution:
 79            return matched_error_resolution
 80
 81        return create_fallback_error_resolution(response_or_exception)
 82
 83    @property
 84    def backoff_strategies(self) -> Optional[List[BackoffStrategy]]:
 85        """
 86        Combines backoff strategies from all child error handlers into a single flattened list.
 87
 88        When used with HttpRequester, note the following behavior:
 89        - In HttpRequester.__post_init__, the entire list of backoff strategies is assigned to the error handler
 90        - However, the error handler's backoff_time() method only ever uses the first non-None strategy in the list
 91        - This means that if any backoff strategies are present, the first non-None strategy becomes the default
 92        - This applies to both user-defined response filters and errors from DEFAULT_ERROR_MAPPING
 93        - The list structure is not used to map different strategies to different error conditions
 94        - Therefore, subsequent strategies in the list will not be used
 95
 96        Returns None if no handlers have strategies defined, which will result in HttpRequester using its default backoff strategy.
 97        """
 98        all_strategies = []
 99        for handler in self.error_handlers:
100            if hasattr(handler, "backoff_strategies") and handler.backoff_strategies:
101                all_strategies.extend(handler.backoff_strategies)
102        return all_strategies if all_strategies else None

 27@dataclass
 28class DefaultErrorHandler(ErrorHandler):
 29    """
 30    Default error handler.
 31
 32    By default, the handler will only use the `DEFAULT_ERROR_MAPPING` that is part of the Python CDK's `HttpStatusErrorHandler`.
 33
 34    If the response is successful, then a SUCCESS_RESOLUTION is returned.
 35    Otherwise, iterate over the response_filters.
 36    If any of the filter match the response, then return the appropriate status.
 37    When `DefaultErrorHandler.backoff_time()` is invoked, iterate sequentially over the backoff_strategies and return the first non-None backoff time, else return None.
 38
 39    Sample configs:
 40
 41    1. retry 10 times
 42    `
 43        error_handler:
 44          max_retries: 10
 45    `
 46    2. backoff for 5 seconds
 47    `
 48        error_handler:
 49          backoff_strategies:
 50            - type: "ConstantBackoff"
 51              backoff_time_in_seconds: 5
 52    `
 53    3. retry on HTTP 404
 54    `
 55        error_handler:
 56          response_filters:
 57            - http_codes: [ 404 ]
 58              action: RETRY
 59    `
 60    4. ignore HTTP 404
 61    `
 62      error_handler:
 63        response_filters:
 64          - http_codes: [ 404 ]
 65            action: IGNORE
 66    `
 67    5. retry if error message contains `retrythisrequest!` substring
 68    `
 69        error_handler:
 70          response_filters:
 71            - error_message_contain: "retrythisrequest!"
 72              action: IGNORE
 73    `
 74    6. retry if 'code' is a field present in the response body
 75    `
 76        error_handler:
 77          response_filters:
 78            - predicate: "{{ 'code' in response }}"
 79              action: IGNORE
 80    `
 81
 82    7. ignore 429 and retry on 404
 83    `
 84        error_handler:
 85        - http_codes: [ 429 ]
 86          action: IGNORE
 87        - http_codes: [ 404 ]
 88          action: RETRY
 89    `
 90
 91    Attributes:
 92        response_filters (Optional[List[HttpResponseFilter]]): response filters to iterate on
 93        max_retries (Optional[Union[int, str]]): maximum retry attempts. Either a hardcoded int or
 94            a string that interpolates from the connector config (e.g.
 95            `"{{ config['max_retries_on_throttle'] }}"`). The string variant is evaluated once at
 96            construction time and replaced with the resolved int.
 97        backoff_strategies (Optional[List[BackoffStrategy]]): list of backoff strategies to use to determine how long
 98        to wait before retrying
 99    """
100
101    parameters: InitVar[Mapping[str, Any]]
102    config: Config
103    response_filters: Optional[List[HttpResponseFilter]] = None
104    # The base class declares max_retries as Optional[int]. We widen the input type to
105    # also accept a Jinja-interpolatable string (e.g. "{{ config['max_retries_on_throttle'] }}"),
106    # which is resolved to an int in __post_init__ so the post-construction invariant matches
107    # the base class contract.
108    max_retries: Optional[Union[int, str]] = 5  # type: ignore[assignment]
109    max_time: int = 60 * 10
110    backoff_strategies: Optional[List[BackoffStrategy]] = None
111
112    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
113        if not self.response_filters:
114            self.response_filters = [HttpResponseFilter(config=self.config, parameters={})]
115
116        self._last_request_to_attempt_count: MutableMapping[requests.PreparedRequest, int] = {}
117
118        if isinstance(self.max_retries, str):
119            evaluated = InterpolatedString(
120                string=self.max_retries, default="5", parameters=parameters
121            ).eval(config=self.config)
122            try:
123                self.max_retries = int(evaluated)
124            except (TypeError, ValueError) as exc:
125                raise ValueError(
126                    f"DefaultErrorHandler.max_retries did not evaluate to an integer "
127                    f"(got {evaluated!r})"
128                ) from exc
129
130    def interpret_response(
131        self, response_or_exception: Optional[Union[requests.Response, Exception]]
132    ) -> ErrorResolution:
133        if self.response_filters:
134            for response_filter in self.response_filters:
135                matched_error_resolution = response_filter.matches(
136                    response_or_exception=response_or_exception
137                )
138                if matched_error_resolution:
139                    return matched_error_resolution
140        if isinstance(response_or_exception, requests.Response):
141            if response_or_exception.ok:
142                return SUCCESS_RESOLUTION
143
144        default_reponse_filter = DefaultHttpResponseFilter(parameters={}, config=self.config)
145        default_response_filter_resolution = default_reponse_filter.matches(response_or_exception)
146
147        return (
148            default_response_filter_resolution
149            if default_response_filter_resolution
150            else create_fallback_error_resolution(response_or_exception)
151        )
152
153    def backoff_time(
154        self,
155        response_or_exception: Optional[Union[requests.Response, requests.RequestException]],
156        attempt_count: int = 0,
157    ) -> Optional[float]:
158        backoff = None
159        if self.backoff_strategies:
160            for backoff_strategy in self.backoff_strategies:
161                backoff = backoff_strategy.backoff_time(
162                    response_or_exception=response_or_exception, attempt_count=attempt_count
163                )
164                if backoff:
165                    return backoff
166        return backoff

12class ErrorHandler(ABC):
13    """
14    Abstract base class to determine how to handle a failed HTTP request.
15    """
16
17    @property
18    @abstractmethod
19    def max_retries(self) -> Optional[int]:
20        """
21        The maximum number of retries to attempt before giving up.
22        """
23        pass
24
25    @property
26    @abstractmethod
27    def max_time(self) -> Optional[int]:
28        """
29        The maximum amount of time in seconds to retry before giving up.
30        """
31        pass
32
33    @abstractmethod
34    def interpret_response(
35        self, response: Optional[Union[requests.Response, Exception]]
36    ) -> ErrorResolution:
37        """
38        Interpret the response or exception and return the corresponding response action, failure type, and error message.
39
40        :param response: The HTTP response object or exception raised during the request.
41        :return: A tuple containing the response action, failure type, and error message.
42        """
43        pass

 25@dataclass
 26class HttpResponseFilter:
 27    """
 28    Filter to select a response based on its HTTP status code, error message or a predicate.
 29    If a response matches the filter, the response action, failure_type, and error message are returned as an ErrorResolution object.
 30    For http_codes declared in the filter, the failure_type will default to `system_error`.
 31    To override default failure_type use configured failure_type with ResponseAction.FAIL.
 32
 33    Attributes:
 34        action (Union[ResponseAction, str]): action to execute if a request matches
 35        failure_type (Union[ResponseAction, str]): failure type of traced exception if a response matches the filter
 36        http_codes (Set[int]): http code of matching requests
 37        error_message_contains (str): error substring of matching requests
 38        predicate (str): predicate to apply to determine if a request is matching
 39        error_message (Union[InterpolatedString, str): error message to display if the response matches the filter
 40    """
 41
 42    config: Config
 43    parameters: InitVar[Mapping[str, Any]]
 44    action: Optional[Union[ResponseAction, str]] = None
 45    failure_type: Optional[Union[FailureType, str]] = None
 46    http_codes: Optional[Set[int]] = None
 47    error_message_contains: Optional[str] = None
 48    predicate: Union[InterpolatedBoolean, str] = ""
 49    error_message: Union[InterpolatedString, str] = ""
 50
 51    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 52        if self.action is not None:
 53            if (
 54                self.http_codes is None
 55                and self.predicate is None
 56                and self.error_message_contains is None
 57            ):
 58                raise ValueError(
 59                    "HttpResponseFilter requires a filter condition if an action is specified"
 60                )
 61            elif isinstance(self.action, str):
 62                self.action = ResponseAction[self.action]
 63        self.http_codes = self.http_codes or set()
 64        if isinstance(self.predicate, str):
 65            self.predicate = InterpolatedBoolean(condition=self.predicate, parameters=parameters)
 66        self.error_message = InterpolatedString.create(
 67            string_or_interpolated=self.error_message, parameters=parameters
 68        )
 69        self._error_message_parser = JsonErrorMessageParser()
 70        if self.failure_type and isinstance(self.failure_type, str):
 71            self.failure_type = FailureType[self.failure_type]
 72
 73    def matches(
 74        self, response_or_exception: Optional[Union[requests.Response, Exception]]
 75    ) -> Optional[ErrorResolution]:
 76        filter_action = self._matches_filter(response_or_exception)
 77        mapped_key = (
 78            response_or_exception.status_code
 79            if isinstance(response_or_exception, requests.Response)
 80            else response_or_exception.__class__
 81        )
 82
 83        if isinstance(mapped_key, (int, Exception)):
 84            default_mapped_error_resolution = self._match_default_error_mapping(mapped_key)
 85        else:
 86            default_mapped_error_resolution = None
 87
 88        if filter_action is not None:
 89            default_error_message = (
 90                default_mapped_error_resolution.error_message
 91                if default_mapped_error_resolution
 92                else ""
 93            )
 94            error_message = None
 95            if isinstance(response_or_exception, requests.Response):
 96                error_message = self._create_error_message(response_or_exception)
 97            error_message = error_message or default_error_message
 98
 99            if self.failure_type and filter_action == ResponseAction.FAIL:
100                failure_type = self.failure_type
101            elif default_mapped_error_resolution:
102                failure_type = default_mapped_error_resolution.failure_type
103            else:
104                failure_type = FailureType.system_error
105
106            return ErrorResolution(
107                response_action=filter_action,
108                failure_type=failure_type,
109                error_message=error_message,
110            )
111
112        if (
113            (isinstance(self.http_codes, list) and len(self.http_codes)) is None
114            and self.predicate is None
115            and self.error_message_contains is None
116        ) and default_mapped_error_resolution:
117            return default_mapped_error_resolution
118
119        return None
120
121    def _match_default_error_mapping(
122        self, mapped_key: Union[int, type[Exception]]
123    ) -> Optional[ErrorResolution]:
124        return DEFAULT_ERROR_MAPPING.get(mapped_key)
125
126    def _matches_filter(
127        self, response_or_exception: Optional[Union[requests.Response, Exception]]
128    ) -> Optional[ResponseAction]:
129        """
130        Apply the HTTP filter on the response and return the action to execute if it matches
131        :param response: The HTTP response to evaluate
132        :return: The action to execute. None if the response does not match the filter
133        """
134        if isinstance(response_or_exception, requests.Response) and (
135            response_or_exception.status_code in self.http_codes  # type: ignore # http_codes set is always initialized to a value in __post_init__
136            or self._response_matches_predicate(response_or_exception)
137            or self._response_contains_error_message(response_or_exception)
138        ):
139            return self.action  # type: ignore # action is always cast to a ResponseAction not a str
140        return None
141
142    @staticmethod
143    def _safe_response_json(response: requests.Response) -> dict[str, Any]:
144        try:
145            return response.json()  # type: ignore # Response.json() returns a dictionary even if the signature does not
146        except requests.exceptions.JSONDecodeError:
147            return {}
148
149    def _create_error_message(self, response: requests.Response) -> Optional[str]:
150        """
151        Construct an error message based on the specified message template of the filter.
152        :param response: The HTTP response which can be used during interpolation
153        :return: The evaluated error message string to be emitted
154        """
155        return self.error_message.eval(  # type: ignore[no-any-return, union-attr]
156            self.config, response=self._safe_response_json(response), headers=response.headers
157        )
158
159    def _response_matches_predicate(self, response: requests.Response) -> bool:
160        return (
161            bool(
162                self.predicate.condition  # type:ignore[union-attr]
163                and self.predicate.eval(  # type:ignore[union-attr]
164                    None,  # type: ignore[arg-type]
165                    response=self._safe_response_json(response),
166                    headers=response.headers,
167                )
168            )
169            if self.predicate
170            else False
171        )
172
173    def _response_contains_error_message(self, response: requests.Response) -> bool:
174        if not self.error_message_contains:
175            return False
176        else:
177            error_message = self._error_message_parser.parse_response_error_message(
178                response=response
179            )
180            return bool(error_message and self.error_message_contains in error_message)