airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies

17@dataclass
18class ConstantBackoffStrategy(BackoffStrategy):
19    """
20    Backoff strategy with a constant backoff interval
21
22    Attributes:
23        backoff_time_in_seconds (float): time to backoff before retrying a retryable request.
24    """
25
26    backoff_time_in_seconds: Union[float, InterpolatedString, str]
27    parameters: InitVar[Mapping[str, Any]]
28    config: Config
29    jitter_range_in_seconds: Optional[float] = None
30
31    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
32        if not isinstance(self.backoff_time_in_seconds, InterpolatedString):
33            self.backoff_time_in_seconds = str(self.backoff_time_in_seconds)
34        if isinstance(self.backoff_time_in_seconds, float):
35            self.backoff_time_in_seconds = InterpolatedString.create(
36                str(self.backoff_time_in_seconds), parameters=parameters
37            )
38        else:
39            self.backoff_time_in_seconds = InterpolatedString.create(
40                self.backoff_time_in_seconds, parameters=parameters
41            )
42
43    def backoff_time(
44        self,
45        response_or_exception: Optional[Union[requests.Response, requests.RequestException]],
46        attempt_count: int,
47    ) -> Optional[float]:
48        backoff_time = float(
49            cast(InterpolatedString, self.backoff_time_in_seconds).eval(self.config)
50        )
51        if self.jitter_range_in_seconds is None:
52            return backoff_time
53
54        return random.uniform(backoff_time, backoff_time + (self.jitter_range_in_seconds * 2))

17@dataclass
18class ExponentialBackoffStrategy(BackoffStrategy):
19    """
20    Backoff strategy with an exponential backoff interval
21
22    Attributes:
23        factor (float): multiplicative factor
24    """
25
26    parameters: InitVar[Mapping[str, Any]]
27    config: Config
28    factor: Union[float, InterpolatedString, str] = 5
29    jitter_range_in_seconds: Optional[float] = None
30
31    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
32        if not isinstance(self.factor, InterpolatedString):
33            self.factor = str(self.factor)
34        if isinstance(self.factor, float):
35            self._factor = InterpolatedString.create(str(self.factor), parameters=parameters)
36        else:
37            self._factor = InterpolatedString.create(self.factor, parameters=parameters)
38
39    @property
40    def _retry_factor(self) -> float:
41        return float(self._factor.eval(self.config))
42
43    def backoff_time(
44        self,
45        response_or_exception: Optional[Union[requests.Response, requests.RequestException]],
46        attempt_count: int,
47    ) -> Optional[float]:
48        backoff_time = float(self._retry_factor * 2**attempt_count)
49        if self.jitter_range_in_seconds is None:
50            return backoff_time
51
52        return random.uniform(backoff_time, backoff_time + (self.jitter_range_in_seconds * 2))

 28@dataclass
 29class WaitTimeFromHeaderBackoffStrategy(BackoffStrategy):
 30    """
 31    Extract wait time from http header
 32
 33    Attributes:
 34        header (str): header to read wait time from
 35        regex (Optional[str]): optional regex to apply on the header to extract its value
 36        max_waiting_time_in_seconds (Optional[Union[float, InterpolatedString, str]]): stop the stream
 37            rather than wait this long or longer -- the bound is inclusive, so a wait exactly
 38            equal to it is refused. Only governs waits that are actually taken: on a
 39            rate-limited response where the authenticator holds another credential with quota,
 40            `HttpClient` rotates onto it instead of asking this strategy for a wait, and the
 41            bound does not apply. Any other retryable error still consults this strategy.
 42    """
 43
 44    header: Union[InterpolatedString, str]
 45    parameters: InitVar[Mapping[str, Any]]
 46    config: Config
 47    regex: Optional[Union[InterpolatedString, str]] = None
 48    max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None
 49
 50    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 51        self.regex = (
 52            InterpolatedString.create(self.regex, parameters=parameters) if self.regex else None
 53        )
 54        self.header = InterpolatedString.create(self.header, parameters=parameters)
 55        self._max_waiting_time_in_seconds = interpolated_max_waiting_time(
 56            self.max_waiting_time_in_seconds, parameters
 57        )
 58        # Resolved here rather than only at the first retryable error. `config` is a field and this
 59        # cap interpolates over `config` alone, so it is fully knowable the moment the component
 60        # exists -- and since `HttpClient` decides token rotation before it asks a strategy for a
 61        # wait, a cap that cannot be evaluated would otherwise stay silent for as long as a spare
 62        # credential keeps the strategies from running. A manifest mistake belongs at startup.
 63        evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config)
 64
 65    def backoff_time(
 66        self,
 67        response_or_exception: Optional[Union[requests.Response, requests.RequestException]],
 68        attempt_count: int,
 69    ) -> Optional[float]:
 70        header = self.header.eval(config=self.config)  # type: ignore  # header is always cast to an interpolated stream
 71        if self.regex:
 72            evaled_regex = self.regex.eval(self.config)  # type: ignore # header is always cast to an interpolated string
 73            regex = re.compile(evaled_regex)
 74        else:
 75            regex = None
 76        header_value = None
 77        if isinstance(response_or_exception, requests.Response):
 78            header_value = get_numeric_value_from_header(response_or_exception, header, regex)
 79            max_waiting_time = evaluate_max_waiting_time(
 80                self._max_waiting_time_in_seconds, self.config
 81            )
 82            # Not always reached: `HttpClient` decides token rotation before it asks a strategy
 83            # for a wait, so on a rate limit where the authenticator has another credential with
 84            # quota this check does not run. The cap bounds waiting, and that path is not
 85            # waiting.
 86            # `max_waiting_time is not None` rather than a truthiness check, so that 0 means
 87            # "never wait" instead of silently disabling the cap. The comparison stays `>=`,
 88            # which is what this cap has always done, and `WaitUntilTimeFromHeader` matches it --
 89            # a wait exactly equal to the cap is refused by both.
 90            # `header_value` is checked for truthiness rather than `is not None` on purpose: a
 91            # header of `0` asks for no wait at all, which no cap -- not even 0 -- should refuse.
 92            if max_waiting_time is not None and header_value and header_value >= max_waiting_time:
 93                raise AirbyteTracedException(
 94                    internal_message=(
 95                        f"Rate limit wait time {header_value}s is greater than or equal to the "
 96                        f"maximum of {max_waiting_time}s this stream is allowed to wait. "
 97                        f"Stopping the stream..."
 98                    ),
 99                    message="The rate limit wait time is longer than the connector is allowed to wait.",
100                    failure_type=FailureType.transient_error,
101                )
102        return header_value

 30@dataclass
 31class WaitTimeFromResponseBackoffStrategy(BackoffStrategy):
 32    """
 33    Extract wait time, in relative seconds, from the JSON response body or the response headers.
 34
 35    Attributes:
 36        wait_time_in_seconds: Expression defining how long to wait before retrying. An empty,
 37            zero, negative, non-finite or non-numeric result, or an expression that cannot be
 38            evaluated on the response body, defers to the next backoff strategy.
 39        max_waiting_time_in_seconds: Optional inclusive cap on the computed wait.
 40    """
 41
 42    wait_time_in_seconds: Union[InterpolatedString, str]
 43    parameters: InitVar[Mapping[str, Any]]
 44    config: Config
 45    max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None
 46
 47    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 48        # Evaluated with `JinjaInterpolation` directly rather than `InterpolatedString`: the latter
 49        # caches a template as plain text when its first evaluation returns the template unchanged,
 50        # which is what `JinjaInterpolation` does when rendering raises `TypeError`. One response
 51        # body of an unexpected shape would then turn the strategy off for the rest of the sync.
 52        self._wait_time_template = (
 53            self.wait_time_in_seconds.string
 54            if isinstance(self.wait_time_in_seconds, InterpolatedString)
 55            else self.wait_time_in_seconds
 56        )
 57        self._interpolation = JinjaInterpolation()
 58        self._parameters = parameters
 59        undeclared = (
 60            JinjaInterpolation._find_undeclared_variables(self._wait_time_template)
 61            - _INTERPOLATION_CONTEXT
 62        )
 63        if undeclared:
 64            # A typo such as `resposne` would otherwise fail on every response and leave the
 65            # strategy silently inert; it is a manifest bug, so it fails at startup.
 66            raise AirbyteTracedException(
 67                internal_message=(
 68                    f"WaitTimeFromResponse `wait_time_in_seconds` {self._wait_time_template!r} "
 69                    f"uses undeclared variables {sorted(undeclared)}; only "
 70                    f"{sorted(_INTERPOLATION_CONTEXT)} are available."
 71                ),
 72                message=(
 73                    "The connector's rate limit wait time expression is invalid. This is a "
 74                    "problem with the connector rather than with your configuration."
 75                ),
 76                failure_type=FailureType.system_error,
 77            )
 78        self._max_waiting_time_in_seconds = interpolated_max_waiting_time(
 79            self.max_waiting_time_in_seconds, parameters
 80        )
 81        evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config)
 82
 83    def backoff_time(
 84        self,
 85        response_or_exception: Optional[Union[requests.Response, requests.RequestException]],
 86        attempt_count: int,
 87    ) -> Optional[float]:
 88        response: Any = {}
 89        headers: Mapping[str, Any] = {}
 90        if isinstance(response_or_exception, requests.Response):
 91            try:
 92                response = response_or_exception.json()
 93            except (ValueError, requests.RequestException, RecursionError):
 94                # Unreadable body (not JSON, too deeply nested, or a broken stream): interpolate
 95                # with an empty response so an expression reading only `headers` still works.
 96                pass
 97            headers = response_or_exception.headers
 98
 99        try:
100            evaluated_wait_time = self._interpolation.eval(
101                self._wait_time_template,
102                self.config,
103                # The template as its own default re-raises an `UndefinedError` from the first
104                # render instead of returning None, so it is reported below.
105                self._wait_time_template,
106                parameters=self._parameters,
107                response=response,
108                headers=headers,
109            )
110            if evaluated_wait_time == self._wait_time_template:
111                # `JinjaInterpolation._eval` returns the raw template when rendering raises
112                # `TypeError`, e.g. arithmetic on a string from the body.
113                raise TypeError("the expression could not be rendered")
114        except (UndefinedError, TypeError, ValueError, ArithmeticError) as error:
115            # The strategy runs on every retryable response, not only the one the expression was
116            # written for: a body of another shape (a list, or a string where an object was
117            # expected) must defer to the next strategy, not fail the stream. The body is not
118            # logged; the error type is enough to point at the expression.
119            LOGGER.warning(
120                "WaitTimeFromResponse: `wait_time_in_seconds` could not be evaluated on this "
121                f"response ({type(error).__name__}); deferring to the next backoff strategy."
122            )
123            return None
124        if evaluated_wait_time is None or (
125            isinstance(evaluated_wait_time, str) and not evaluated_wait_time.strip()
126        ):
127            return None
128
129        try:
130            wait_time = float(evaluated_wait_time)
131        except (TypeError, ValueError, ArithmeticError):
132            # Mirrors the header strategies, which return None for a value they cannot read, so
133            # the next backoff strategy decides. The value is not logged: it comes from the
134            # response body and may hold anything.
135            LOGGER.warning(
136                "WaitTimeFromResponse: `wait_time_in_seconds` resolved to a non-numeric "
137                f"{type(evaluated_wait_time).__name__}; deferring to the next backoff strategy."
138            )
139            return None
140        if not math.isfinite(wait_time) or wait_time < 0:
141            # A retry time that has already passed gives a negative wait, which is not a wait to
142            # take: the next strategy decides, as `WaitUntilTimeFromHeader` does for a past
143            # timestamp. Returned as-is, a negative value would reach `time.sleep` and fail the
144            # stream. A zero wait is returned as 0, but `DefaultErrorHandler` treats it as no
145            # answer and moves on to the next strategy, as it does for `WaitTimeFromHeader`.
146            return None
147        max_waiting_time = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config)
148        if max_waiting_time is not None and wait_time and wait_time >= max_waiting_time:
149            raise AirbyteTracedException(
150                internal_message=(
151                    f"Rate limit wait time {wait_time}s is greater than or equal to the "
152                    f"maximum of {max_waiting_time}s this stream is allowed to wait. "
153                    f"Stopping the stream..."
154                ),
155                message="The rate limit wait time is longer than the connector is allowed to wait.",
156                failure_type=FailureType.transient_error,
157            )
158
159        return wait_time

 29@dataclass
 30class WaitUntilTimeFromHeaderBackoffStrategy(BackoffStrategy):
 31    """
 32    Extract time at which we can retry the request from response header
 33    and wait for the difference between now and that time
 34
 35    Attributes:
 36        header (str): header to read wait time from
 37        min_wait (Optional[Union[float, InterpolatedString, str]]): minimum time to wait for safety
 38        regex (Optional[str]): optional regex to apply on the header to extract its value
 39        max_waiting_time_in_seconds (Optional[Union[float, InterpolatedString, str]]): stop the stream
 40            rather than wait this long or longer -- the bound is inclusive, so a wait exactly
 41            equal to it is refused. Only governs waits that are actually taken: on a
 42            rate-limited response where the authenticator holds another credential with quota,
 43            `HttpClient` rotates onto it instead of asking this strategy for a wait, and the
 44            bound does not apply. Any other retryable error still consults this strategy.
 45    """
 46
 47    header: Union[InterpolatedString, str]
 48    parameters: InitVar[Mapping[str, Any]]
 49    config: Config
 50    min_wait: Optional[Union[float, InterpolatedString, str]] = None
 51    regex: Optional[Union[InterpolatedString, str]] = None
 52    max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None
 53
 54    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 55        self.header = InterpolatedString.create(self.header, parameters=parameters)
 56        self.regex = (
 57            InterpolatedString.create(self.regex, parameters=parameters) if self.regex else None
 58        )
 59        if not isinstance(self.min_wait, InterpolatedString):
 60            self.min_wait = InterpolatedString.create(str(self.min_wait), parameters=parameters)
 61        self._max_waiting_time_in_seconds = interpolated_max_waiting_time(
 62            self.max_waiting_time_in_seconds, parameters
 63        )
 64        # Resolved here rather than only at the first retryable error. `config` is a field and this
 65        # cap interpolates over `config` alone, so it is fully knowable the moment the component
 66        # exists -- and since `HttpClient` decides token rotation before it asks a strategy for a
 67        # wait, a cap that cannot be evaluated would otherwise stay silent for as long as a spare
 68        # credential keeps the strategies from running. A manifest mistake belongs at startup.
 69        evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config)
 70
 71    def backoff_time(
 72        self,
 73        response_or_exception: Optional[Union[requests.Response, requests.RequestException]],
 74        attempt_count: int,
 75    ) -> Optional[float]:
 76        now = time.time()
 77        header = self.header.eval(self.config)  # type: ignore # header is always cast to an interpolated string
 78        if self.regex:
 79            evaled_regex = self.regex.eval(self.config)  # type: ignore # header is always cast to an interpolated string
 80            regex = re.compile(evaled_regex)
 81        else:
 82            regex = None
 83        wait_until = None
 84        if isinstance(response_or_exception, requests.Response):
 85            # get_numeric_value_from_header returns a float or None, never a string
 86            wait_until = get_numeric_value_from_header(response_or_exception, header, regex)
 87        min_wait = self.min_wait.eval(self.config)  # type: ignore # header is always cast to an interpolated string
 88        if not wait_until:
 89            return self._capped(float(min_wait)) if min_wait else None
 90        wait_time = wait_until - now
 91        if min_wait:
 92            return self._capped(float(max(wait_time, min_wait)))
 93        elif wait_time < 0:
 94            return None
 95        return self._capped(wait_time)
 96
 97    def _capped(self, wait_time: float) -> float:
 98        """Raise rather than wait `max_waiting_time_in_seconds` or longer.
 99
100        The cap is compared against the wait this strategy is about to return, not against the
101        raw header: unlike `Retry-After`, the header here is an absolute timestamp, so only the
102        computed difference is a duration. It is also applied after the `min_wait` floor, so a
103        cap below the floor wins -- a caller asking never to wait more than N seconds means it,
104        even when the floor would otherwise round the wait up past N.
105
106        Not always reached: `HttpClient` decides token rotation before it asks a strategy for a
107        wait, so on a rate limit where the authenticator has another credential with quota this
108        method does not run and the cap does not apply. Waiting is what the cap bounds, and that
109        path is not waiting.
110        """
111        max_waiting_time = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config)
112        # `>=` rather than `>` to match WaitTimeFromHeader, so one field name does not mean two
113        # different things depending on which strategy it is written on. A cap of 0 therefore
114        # refuses every wait, which is what "never wait" has to mean.
115        if max_waiting_time is not None and wait_time >= max_waiting_time:
116            raise AirbyteTracedException(
117                internal_message=(
118                    f"Rate limit wait time {wait_time}s is greater than or equal to the maximum "
119                    f"of {max_waiting_time}s this stream is allowed to wait. Stopping the stream..."
120                ),
121                message="The rate limit wait time is longer than the connector is allowed to wait.",
122                failure_type=FailureType.transient_error,
123            )
124        return wait_time