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

 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