airbyte_cdk.sources.streams.http

 99class HttpClient:
100    _DEFAULT_MAX_RETRY: int = 5
101    _DEFAULT_MAX_TIME: int = 60 * 10
102    # Backoff used in place of a rate-limit wait when another credential can serve the retry.
103    # Kept non-zero so a misreporting authenticator degrades to a slow retry, not a hot loop
104    # (the retry handler adds a second on top of whatever is returned).
105    TOKEN_ROTATION_BACKOFF: float = 0.1
106    _ACTIONS_TO_RETRY_ON = {
107        ResponseAction.RETRY,
108        ResponseAction.RATE_LIMITED,
109        ResponseAction.REFRESH_TOKEN_THEN_RETRY,
110    }
111
112    def __init__(
113        self,
114        name: str,
115        logger: logging.Logger,
116        error_handler: Optional[ErrorHandler] = None,
117        api_budget: Optional[APIBudget] = None,
118        session: Optional[Union[requests.Session, requests_cache.CachedSession]] = None,
119        authenticator: Optional[AuthBase] = None,
120        use_cache: bool = False,
121        backoff_strategy: Optional[Union[BackoffStrategy, List[BackoffStrategy]]] = None,
122        error_message_parser: Optional[ErrorMessageParser] = None,
123        disable_retries: bool = False,
124        message_repository: Optional[MessageRepository] = None,
125    ):
126        self._name = name
127        self._api_budget: APIBudget = api_budget or APIBudget(policies=[])
128        if session:
129            self._session = session
130        else:
131            self._use_cache = use_cache
132            self._session = self._request_session()
133            self._session.mount(
134                "https://",
135                requests.adapters.HTTPAdapter(
136                    pool_connections=MAX_CONNECTION_POOL_SIZE, pool_maxsize=MAX_CONNECTION_POOL_SIZE
137                ),
138            )
139        if isinstance(authenticator, AuthBase):
140            self._session.auth = authenticator
141        self._logger = logger
142        self._error_handler = error_handler or HttpStatusErrorHandler(self._logger)
143        if backoff_strategy is not None:
144            if isinstance(backoff_strategy, list):
145                self._backoff_strategies = backoff_strategy
146            else:
147                self._backoff_strategies = [backoff_strategy]
148        else:
149            self._backoff_strategies = [DefaultBackoffStrategy()]
150        self._error_message_parser = error_message_parser or JsonErrorMessageParser()
151        self._request_attempt_count: Dict[requests.PreparedRequest, int] = {}
152        self._disable_retries = disable_retries
153        self._message_repository = message_repository
154        self._authenticator_update_failed = False
155
156    @property
157    def cache_filename(self) -> str:
158        """
159        Override if needed. Return the name of cache file
160        Note that if the environment variable REQUEST_CACHE_PATH is not set, the cache will be in-memory only.
161        """
162        return f"{self._name}.sqlite"
163
164    def _request_session(self) -> requests.Session:
165        """
166        Session factory based on use_cache property and call rate limits (api_budget parameter)
167        :return: instance of request-based session
168        """
169        if self._use_cache:
170            cache_dir = os.getenv(ENV_REQUEST_CACHE_PATH)
171            # Use in-memory cache if cache_dir is not set
172            # This is a non-obvious interface, but it ensures we don't write sql files when running unit tests
173            # Use in-memory cache if cache_dir is not set
174            # This is a non-obvious interface, but it ensures we don't write sql files when running unit tests
175            sqlite_path = (
176                str(Path(cache_dir) / self.cache_filename)
177                if cache_dir
178                else "file::memory:?cache=shared"
179            )
180            # By using `PRAGMA synchronous=OFF` and `PRAGMA journal_mode=WAL`, we reduce the possible occurrences of `database table is locked` errors.
181            # Note that those were blindly added at the same time and one or the other might be sufficient to prevent the issues but we have seen good results with both. Feel free to revisit given more information.
182            # There are strong signals that `fast_save` might create problems but if the sync crashes, we start back from the beginning in terms of sqlite anyway so the impact should be minimal. Signals are:
183            # * https://github.com/requests-cache/requests-cache/commit/7fa89ffda300331c37d8fad7f773348a3b5b0236#diff-f43db4a5edf931647c32dec28ea7557aae4cae8444af4b26c8ecbe88d8c925aaR238
184            # * https://github.com/requests-cache/requests-cache/commit/7fa89ffda300331c37d8fad7f773348a3b5b0236#diff-2e7f95b7d7be270ff1a8118f817ea3e6663cdad273592e536a116c24e6d23c18R164-R168
185            # * `If the application running SQLite crashes, the data will be safe, but the database [might become corrupted](https://www.sqlite.org/howtocorrupt.html#cfgerr) if the operating system crashes or the computer loses power before that data has been written to the disk surface.` in [this description](https://www.sqlite.org/pragma.html#pragma_synchronous).
186            backend = requests_cache.SQLiteCache(sqlite_path, fast_save=True, wal=True)
187            return CachedLimiterSession(
188                cache_name=sqlite_path,
189                backend=backend,
190                api_budget=self._api_budget,
191                match_headers=True,
192            )
193        else:
194            return LimiterSession(api_budget=self._api_budget)
195
196    def clear_cache(self) -> None:
197        """
198        Clear cached requests for current session, can be called any time
199        """
200        if isinstance(self._session, requests_cache.CachedSession):
201            self._session.cache.clear()  # type: ignore # cache.clear is not typed
202
203    def _dedupe_query_params(
204        self, url: str, params: Optional[Mapping[str, str]]
205    ) -> Mapping[str, str]:
206        """
207        Remove query parameters from params mapping if they are already encoded in the URL.
208        :param url: URL with
209        :param params:
210        :return:
211        """
212        if params is None:
213            params = {}
214        query_string = urllib.parse.urlparse(url).query
215        query_dict = {k: v[0] for k, v in urllib.parse.parse_qs(query_string).items()}
216
217        duplicate_keys_with_same_value = {
218            k for k in query_dict.keys() if str(params.get(k)) == str(query_dict[k])
219        }
220        return {k: v for k, v in params.items() if k not in duplicate_keys_with_same_value}
221
222    def _create_prepared_request(
223        self,
224        http_method: str,
225        url: str,
226        dedupe_query_params: bool = False,
227        headers: Optional[Mapping[str, str]] = None,
228        params: Optional[Mapping[str, str]] = None,
229        json: Optional[Mapping[str, Any]] = None,
230        data: Optional[Union[str, Mapping[str, Any]]] = None,
231    ) -> requests.PreparedRequest:
232        if dedupe_query_params:
233            query_params = self._dedupe_query_params(url, params)
234        else:
235            query_params = params or {}
236        args = {"method": http_method, "url": url, "headers": headers, "params": query_params}
237        if http_method.upper() in BODY_REQUEST_METHODS:
238            if json and data:
239                raise RequestBodyException(
240                    "At the same time only one of the 'request_body_data' and 'request_body_json' functions can return data"
241                )
242            elif json:
243                args["json"] = json
244            elif data:
245                args["data"] = data
246        prepared_request: requests.PreparedRequest = self._session.prepare_request(
247            requests.Request(**args)
248        )
249
250        return prepared_request
251
252    @property
253    def _max_retries(self) -> int:
254        """
255        Determines the max retries based on the provided error handler.
256        """
257        max_retries = None
258        if self._disable_retries:
259            max_retries = 0
260        else:
261            max_retries = self._error_handler.max_retries
262        return max_retries if max_retries is not None else self._DEFAULT_MAX_RETRY
263
264    @property
265    def _max_time(self) -> int:
266        """
267        Determines the max time based on the provided error handler.
268        """
269        return (
270            self._error_handler.max_time
271            if self._error_handler.max_time is not None
272            else self._DEFAULT_MAX_TIME
273        )
274
275    def _send_with_retry(
276        self,
277        request: requests.PreparedRequest,
278        request_kwargs: Mapping[str, Any],
279        log_formatter: Optional[Callable[[requests.Response], Any]] = None,
280        exit_on_rate_limit: Optional[bool] = False,
281    ) -> requests.Response:
282        """
283        Sends a request with retry logic.
284
285        Args:
286            request (requests.PreparedRequest): The prepared HTTP request to send.
287            request_kwargs (Mapping[str, Any]): Additional keyword arguments for the request.
288
289        Returns:
290            requests.Response: The HTTP response received from the server after retries.
291        """
292
293        max_retries = self._max_retries
294        max_tries = max(0, max_retries) + 1
295        max_time = self._max_time
296
297        user_backoff_handler = user_defined_backoff_handler(max_tries=max_tries, max_time=max_time)(
298            self._send
299        )
300        rate_limit_backoff_handler = rate_limit_default_backoff_handler(max_tries=max_tries)
301        backoff_handler = http_client_default_backoff_handler(
302            max_tries=max_tries, max_time=max_time
303        )
304        # backoff handlers wrap _send, so it will always return a response -- except when all retries are exhausted
305        try:
306            response = backoff_handler(rate_limit_backoff_handler(user_backoff_handler))(
307                request,
308                request_kwargs,
309                log_formatter=log_formatter,
310                exit_on_rate_limit=exit_on_rate_limit,
311            )  # type: ignore # mypy can't infer that backoff_handler wraps _send
312
313            return response
314        except BaseBackoffException as e:
315            self._logger.error("Retries exhausted with backoff exception.", exc_info=True)
316
317            is_rate_limited = (
318                isinstance(e.response, requests.Response)
319                and e.response.status_code == requests.codes.too_many_requests
320            )
321
322            if is_rate_limited:
323                raise AirbyteTracedException(
324                    internal_message=f"Rate limit retry budget exhausted. Last exception: {e}",
325                    message="API rate limit exceeded.",
326                    failure_type=FailureType.transient_error,
327                    exception=e,
328                    stream_descriptor=StreamDescriptor(name=self._name),
329                )
330
331            raise AirbyteTracedException(
332                internal_message=f"Exhausted available request attempts. Exception: {e}",
333                message=f"Exhausted available request attempts. Please see logs for more details. Exception: {e}",
334                failure_type=e.failure_type or FailureType.system_error,
335                exception=e,
336                stream_descriptor=StreamDescriptor(name=self._name),
337            )
338
339    def _can_retry_on_another_token(self, request: requests.PreparedRequest) -> bool:
340        """Whether the authenticator can serve this request from a different credential now.
341
342        Opted into by implementing `TokenRotatingAuthenticator`.
343        """
344        authenticator = getattr(self._session, "auth", None)
345        if not isinstance(authenticator, TokenRotatingAuthenticator):
346            return False
347        try:
348            return bool(authenticator.has_alternative_token(request))
349        except Exception:
350            # Falling back to the computed wait is always safe, so never fail a retry over this.
351            self._logger.debug(
352                "Authenticator failed to report credential availability", exc_info=True
353            )
354            return False
355
356    def _update_authenticator_from_response(
357        self, request: requests.PreparedRequest, response: requests.Response
358    ) -> None:
359        """Let a quota-tracking authenticator reconcile its state against the server.
360
361        Authenticators only ever see requests, so an authenticator that tracks per-token quota
362        has no way to learn that the server disagrees with its local bookkeeping. This is the
363        feedback channel, mirroring what `LimiterMixin.send` does for the API budget.
364
365        Opted into by implementing `ResponseAwareAuthenticator`.
366        """
367        authenticator = getattr(self._session, "auth", None)
368        if not isinstance(authenticator, ResponseAwareAuthenticator):
369            return
370        if getattr(response, "from_cache", False):
371            # A replayed cached response carries the rate-limit headers from whenever it was
372            # first fetched and consumed no quota of its own.
373            return
374        try:
375            authenticator.update_from_response(request, response)
376        except Exception:
377            # Quota bookkeeping must never turn an otherwise fine response into a failure. Warn
378            # once so a persistently broken update -- which silently degrades the connector back
379            # to single-token behaviour -- is at least diagnosable from default-level logs.
380            if not self._authenticator_update_failed:
381                self._authenticator_update_failed = True
382                self._logger.warning(
383                    "Authenticator failed to update quota state from a response; token rotation "
384                    "may fall back to local counters only. Further occurrences log at debug.",
385                    exc_info=True,
386                )
387            else:
388                self._logger.debug(
389                    "Authenticator failed to update quota state from response", exc_info=True
390                )
391
392    def _send(
393        self,
394        request: requests.PreparedRequest,
395        request_kwargs: Mapping[str, Any],
396        log_formatter: Optional[Callable[[requests.Response], Any]] = None,
397        exit_on_rate_limit: Optional[bool] = False,
398    ) -> requests.Response:
399        if request not in self._request_attempt_count:
400            self._request_attempt_count[request] = 1
401        else:
402            self._request_attempt_count[request] += 1
403            if hasattr(self._session, "auth") and isinstance(self._session.auth, AuthBase):
404                self._session.auth(request)
405
406        self._logger.debug(
407            "Making outbound API request",
408            extra={"headers": request.headers, "url": request.url, "request_body": request.body},
409        )
410
411        response: Optional[requests.Response] = None
412        exc: Optional[requests.RequestException] = None
413
414        try:
415            response = self._session.send(request, **request_kwargs)
416        except requests.RequestException as e:
417            exc = e
418
419        if response is not None:
420            self._update_authenticator_from_response(request, response)
421
422        error_resolution: ErrorResolution = self._error_handler.interpret_response(
423            response if response is not None else exc
424        )
425
426        # Evaluation of response.text can be heavy, for example, if streaming a large response
427        # Do it only in debug mode
428        if self._logger.isEnabledFor(logging.DEBUG) and response is not None:
429            if request_kwargs.get("stream"):
430                self._logger.debug(
431                    "Receiving response, but not logging it as the response is streamed",
432                    extra={"headers": response.headers, "status": response.status_code},
433                )
434            else:
435                self._logger.debug(
436                    "Receiving response",
437                    extra={
438                        "headers": response.headers,
439                        "status": response.status_code,
440                        "body": response.text,
441                    },
442                )
443
444        # Request/response logging for declarative cdk
445        if (
446            log_formatter is not None
447            and response is not None
448            and self._message_repository is not None
449        ):
450            formatter = log_formatter
451            self._message_repository.log_message(
452                Level.DEBUG,
453                lambda: formatter(response),
454            )
455
456        self._handle_error_resolution(
457            response=response,
458            exc=exc,
459            request=request,
460            error_resolution=error_resolution,
461            exit_on_rate_limit=exit_on_rate_limit,
462        )
463
464        return response  # type: ignore # will either return a valid response of type requests.Response or raise an exception
465
466    def _get_response_body(self, response: requests.Response) -> Optional[JsonType]:
467        """
468        Extracts and returns the body of an HTTP response.
469
470        This method attempts to parse the response body as JSON. If the response
471        body is not valid JSON, it falls back to decoding the response content
472        as a UTF-8 string. If both attempts fail, it returns None.
473
474        Args:
475            response (requests.Response): The HTTP response object.
476
477        Returns:
478            Optional[JsonType]: The parsed JSON object as a string, the decoded
479            response content as a string, or None if both parsing attempts fail.
480        """
481        try:
482            return str(response.json())
483        except requests.exceptions.JSONDecodeError:
484            try:
485                return response.content.decode("utf-8")
486            except Exception:
487                return "The Content of the Response couldn't be decoded."
488
489    def _evict_key(self, prepared_request: requests.PreparedRequest) -> None:
490        """
491        Addresses high memory consumption when enabling concurrency in https://github.com/airbytehq/oncall/issues/6821.
492
493        The `_request_attempt_count` attribute keeps growing as multiple requests are made using the same `http_client`.
494        To mitigate this issue, we evict keys for completed requests once we confirm that no further retries are needed.
495        This helps manage memory usage more efficiently while maintaining the necessary logic for retry attempts.
496        """
497        if prepared_request in self._request_attempt_count:
498            del self._request_attempt_count[prepared_request]
499
500    def _handle_error_resolution(
501        self,
502        response: Optional[requests.Response],
503        exc: Optional[requests.RequestException],
504        request: requests.PreparedRequest,
505        error_resolution: ErrorResolution,
506        exit_on_rate_limit: Optional[bool] = False,
507    ) -> None:
508        if error_resolution.response_action not in self._ACTIONS_TO_RETRY_ON:
509            self._evict_key(request)
510
511        if error_resolution.response_action == ResponseAction.RESET_PAGINATION:
512            raise PaginationResetRequiredException()
513
514        # Emit stream status RUNNING with the reason RATE_LIMITED to log that the rate limit has been reached
515        if error_resolution.response_action == ResponseAction.RATE_LIMITED:
516            # TODO: Update to handle with message repository when concurrent message repository is ready
517            reasons = [AirbyteStreamStatusReason(type=AirbyteStreamStatusReasonType.RATE_LIMITED)]
518            message = orjson.dumps(
519                AirbyteMessageSerializer.dump(
520                    stream_status_as_airbyte_message(
521                        StreamDescriptor(name=self._name), AirbyteStreamStatus.RUNNING, reasons
522                    )
523                )
524            ).decode()
525
526            # Simply printing the stream status is a temporary solution and can cause future issues. Currently, the _send method is
527            # wrapped with backoff decorators, and we can only emit messages by iterating record_iterator in the abstract source at the
528            # end of the retry decorator behavior. This approach does not allow us to emit messages in the queue before exiting the
529            # backoff retry loop. Adding `\n` to the message and ignore 'end' ensure that few messages are printed at the same time.
530            print(f"{message}\n", end="", flush=True)
531
532        # Handle REFRESH_TOKEN_THEN_RETRY: Force refresh the OAuth token before retry
533        # This is useful when the API returns 401 but the stored token expiry hasn't been reached yet
534        # Only OAuth authenticators have refresh_and_set_access_token method
535        # Non-OAuth auth types (e.g., BearerAuthenticator) will fall through to normal retry
536        if error_resolution.response_action == ResponseAction.REFRESH_TOKEN_THEN_RETRY:
537            if (
538                hasattr(self._session, "auth")
539                and self._session.auth is not None
540                and hasattr(self._session.auth, "refresh_and_set_access_token")
541            ):
542                try:
543                    self._session.auth.refresh_and_set_access_token()  # type: ignore[union-attr]
544                    self._logger.info(
545                        "Refreshed OAuth token due to REFRESH_TOKEN_THEN_RETRY response action"
546                    )
547                except Exception as refresh_error:
548                    self._logger.warning(
549                        f"Failed to refresh OAuth token: {refresh_error}. Proceeding with retry using existing token."
550                    )
551            else:
552                self._logger.warning(
553                    "REFRESH_TOKEN_THEN_RETRY action received but authenticator does not support token refresh. "
554                    "Proceeding with normal retry."
555                )
556
557        if error_resolution.response_action == ResponseAction.FAIL:
558            if response is not None:
559                filtered_response_message = filter_secrets(
560                    f"Request (body): '{str(request.body)}'. Response (body): '{self._get_response_body(response)}'. Response (headers): '{response.headers}'."
561                )
562                error_message = f"'{request.method}' request to '{request.url}' failed with status code '{response.status_code}' and error message: '{self._error_message_parser.parse_response_error_message(response)}'. {filtered_response_message}"
563            else:
564                error_message = (
565                    f"'{request.method}' request to '{request.url}' failed with exception: '{exc}'"
566                )
567
568            # ensure the exception message is emitted before raised
569            self._logger.error(error_message)
570
571            raise AirbyteTracedException(
572                internal_message=error_message,
573                message=error_resolution.error_message or error_message,
574                failure_type=error_resolution.failure_type,
575            )
576
577        elif error_resolution.response_action == ResponseAction.IGNORE:
578            if response is not None:
579                log_message = f"Ignoring response for '{request.method}' request to '{request.url}' with response code '{response.status_code}'"
580            else:
581                log_message = f"Ignoring response for '{request.method}' request to '{request.url}' with error '{exc}'"
582
583            self._logger.info(error_resolution.error_message or log_message)
584
585        # TODO: Consider dynamic retry count depending on subsequent error codes
586        elif error_resolution.response_action in (
587            ResponseAction.RETRY,
588            ResponseAction.RATE_LIMITED,
589            ResponseAction.REFRESH_TOKEN_THEN_RETRY,
590        ):
591            user_defined_backoff_time = None
592            # Asked before the strategies, not after. The backoff they compute describes the
593            # credential the server just rejected, so when another credential can serve the
594            # retry that wait is irrelevant -- and a strategy is allowed to refuse a wait by
595            # raising (`max_waiting_time_in_seconds`), which would otherwise end the stream
596            # before rotation was ever considered. Rotating is strictly the better outcome
597            # there: it is the same retry, seconds from now, on a credential with quota.
598            #
599            # Two consequences of not calling the strategies, both deliberate. A rate limit
600            # that yields no backoff at all now rotates too, rather than falling through to
601            # the default exponential retry -- on a rotating credential that is the better
602            # behaviour, and `has_alternative_token` only answers True when the retry will
603            # rotate -- the CDK's own authenticator narrows that further, to a sending credential
604            # that is tracked and spent. And a `max_waiting_time_in_seconds` the manifest got
605            # wrong -- one that cannot be evaluated -- is not reported from here, since that
606            # error is raised from inside the strategy. Both capped strategies therefore resolve
607            # the field once in `__post_init__` too, so a manifest mistake fails at startup
608            # rather than waiting for a rate limit that finds no spare credential.
609            rotate_instead_of_waiting = (
610                error_resolution.response_action == ResponseAction.RATE_LIMITED
611                and self._can_retry_on_another_token(request)
612            )
613
614            if rotate_instead_of_waiting:
615                # Says that a wait was skipped without the number, which is no longer computed,
616                # and names the cap explicitly: a connector that configured one gets no other
617                # signal that the retry went ahead without consulting it.
618                self._logger.info(
619                    "Rate limited on the current credential; retrying in "
620                    f"{self.TOKEN_ROTATION_BACKOFF}s with another one instead of waiting for the "
621                    "rate limit to reset. Any configured backoff, including a wait cap, is not "
622                    "evaluated for this retry."
623                )
624                user_defined_backoff_time = self.TOKEN_ROTATION_BACKOFF
625            else:
626                for backoff_strategy in self._backoff_strategies:
627                    backoff_time = backoff_strategy.backoff_time(
628                        response_or_exception=response if response is not None else exc,
629                        attempt_count=self._request_attempt_count[request],
630                    )
631                    if backoff_time:
632                        user_defined_backoff_time = backoff_time
633                        break
634
635            error_message = (
636                error_resolution.error_message
637                or f"Request to {request.url} failed with failure type {error_resolution.failure_type}, response action {error_resolution.response_action}."
638            )
639
640            retry_endlessly = (
641                error_resolution.response_action == ResponseAction.RATE_LIMITED
642                and not exit_on_rate_limit
643            )
644
645            if user_defined_backoff_time:
646                raise UserDefinedBackoffException(
647                    backoff=user_defined_backoff_time,
648                    request=request,
649                    response=(response if response is not None else exc),
650                    error_message=error_message,
651                    failure_type=error_resolution.failure_type,
652                )
653
654            elif retry_endlessly:
655                raise RateLimitBackoffException(
656                    request=request,
657                    response=(response if response is not None else exc),
658                    error_message=error_message,
659                    failure_type=error_resolution.failure_type,
660                )
661
662            raise DefaultBackoffException(
663                request=request,
664                response=(response if response is not None else exc),
665                error_message=error_message,
666                failure_type=error_resolution.failure_type,
667            )
668
669        elif response:
670            try:
671                response.raise_for_status()
672            except requests.HTTPError as e:
673                self._logger.error(response.text)
674                raise e
675
676    @property
677    def name(self) -> str:
678        return self._name
679
680    def send_request(
681        self,
682        http_method: str,
683        url: str,
684        request_kwargs: Mapping[str, Any],
685        headers: Optional[Mapping[str, str]] = None,
686        params: Optional[Mapping[str, str]] = None,
687        json: Optional[Mapping[str, Any]] = None,
688        data: Optional[Union[str, Mapping[str, Any]]] = None,
689        dedupe_query_params: bool = False,
690        log_formatter: Optional[Callable[[requests.Response], Any]] = None,
691        exit_on_rate_limit: Optional[bool] = False,
692    ) -> Tuple[requests.PreparedRequest, requests.Response]:
693        """
694        Prepares and sends request and return request and response objects.
695        """
696
697        request: requests.PreparedRequest = self._create_prepared_request(
698            http_method=http_method,
699            url=url,
700            dedupe_query_params=dedupe_query_params,
701            headers=headers,
702            params=params,
703            json=json,
704            data=data,
705        )
706
707        env_settings = self._session.merge_environment_settings(
708            url=request.url,
709            proxies=request_kwargs.get("proxies", {}),
710            stream=request_kwargs.get("stream"),
711            verify=request_kwargs.get("verify"),
712            cert=request_kwargs.get("cert"),
713        )
714        request_kwargs = {**request_kwargs, **env_settings}
715
716        response: requests.Response = self._send_with_retry(
717            request=request,
718            request_kwargs=request_kwargs,
719            log_formatter=log_formatter,
720            exit_on_rate_limit=exit_on_rate_limit,
721        )
722
723        return request, response

 45class HttpStream(Stream, CheckpointMixin, ABC):
 46    """
 47    Base abstract class for an Airbyte Stream using the HTTP protocol. Basic building block for users building an Airbyte source for a HTTP API.
 48    """
 49
 50    source_defined_cursor = True  # Most HTTP streams use a source defined cursor (i.e: the user can't configure it like on a SQL table)
 51    page_size: Optional[int] = (
 52        None  # Use this variable to define page size for API http requests with pagination support
 53    )
 54
 55    def __init__(
 56        self, authenticator: Optional[AuthBase] = None, api_budget: Optional[APIBudget] = None
 57    ):
 58        self._exit_on_rate_limit: bool = False
 59        self._http_client = HttpClient(
 60            name=self.name,
 61            logger=self.logger,
 62            error_handler=self.get_error_handler(),
 63            api_budget=api_budget or APIBudget(policies=[]),
 64            authenticator=authenticator,
 65            use_cache=self.use_cache,
 66            backoff_strategy=self.get_backoff_strategy(),
 67            message_repository=InMemoryMessageRepository(),
 68        )
 69
 70        # There are three conditions that dictate if RFR should automatically be applied to a stream
 71        # 1. Streams that explicitly initialize their own cursor should defer to it and not automatically apply RFR
 72        # 2. Streams with at least one cursor_field are incremental and thus a superior sync to RFR.
 73        # 3. Streams overriding read_records() do not guarantee that they will call the parent implementation which can perform
 74        #    per-page checkpointing so RFR is only supported if a stream use the default `HttpStream.read_records()` method
 75        if (
 76            not self.cursor
 77            and len(self.cursor_field) == 0
 78            and type(self).read_records is HttpStream.read_records
 79        ):
 80            self.cursor = ResumableFullRefreshCursor()
 81
 82    @property
 83    def exit_on_rate_limit(self) -> bool:
 84        """
 85        :return: False if the stream will retry endlessly when rate limited
 86        """
 87        return self._exit_on_rate_limit
 88
 89    @exit_on_rate_limit.setter
 90    def exit_on_rate_limit(self, value: bool) -> None:
 91        self._exit_on_rate_limit = value
 92
 93    @property
 94    def cache_filename(self) -> str:
 95        """
 96        Override if needed. Return the name of cache file
 97        Note that if the environment variable REQUEST_CACHE_PATH is not set, the cache will be in-memory only.
 98        """
 99        return f"{self.name}.sqlite"
100
101    @property
102    def use_cache(self) -> bool:
103        """
104        Override if needed. If True, all records will be cached.
105        Note that if the environment variable REQUEST_CACHE_PATH is not set, the cache will be in-memory only.
106        """
107        return False
108
109    @property
110    @abstractmethod
111    def url_base(self) -> str:
112        """
113        :return: URL base for the  API endpoint e.g: if you wanted to hit https://myapi.com/v1/some_entity then this should return "https://myapi.com/v1/"
114        """
115
116    @property
117    def http_method(self) -> str:
118        """
119        Override if needed. See get_request_data/get_request_json if using POST/PUT/PATCH.
120        """
121        return "GET"
122
123    @property
124    @deprecated(
125        "Deprecated as of CDK version 3.0.0. "
126        "You should set error_handler explicitly in HttpStream.get_error_handler() instead."
127    )
128    def raise_on_http_errors(self) -> bool:
129        """
130        Override if needed. If set to False, allows opting-out of raising HTTP code exception.
131        """
132        return True
133
134    @property
135    @deprecated(
136        "Deprecated as of CDK version 3.0.0. "
137        "You should set backoff_strategies explicitly in HttpStream.get_backoff_strategy() instead."
138    )
139    def max_retries(self) -> Union[int, None]:
140        """
141        Override if needed. Specifies maximum amount of retries for backoff policy. Return None for no limit.
142        """
143        return 5
144
145    @property
146    @deprecated(
147        "Deprecated as of CDK version 3.0.0. "
148        "You should set backoff_strategies explicitly in HttpStream.get_backoff_strategy() instead."
149    )
150    def max_time(self) -> Union[int, None]:
151        """
152        Override if needed. Specifies maximum total waiting time (in seconds) for backoff policy. Return None for no limit.
153        """
154        return 60 * 10
155
156    @property
157    @deprecated(
158        "Deprecated as of CDK version 3.0.0. "
159        "You should set backoff_strategies explicitly in HttpStream.get_backoff_strategy() instead."
160    )
161    def retry_factor(self) -> float:
162        """
163        Override if needed. Specifies factor for backoff policy.
164        """
165        return 5
166
167    @abstractmethod
168    def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]:
169        """
170        Override this method to define a pagination strategy.
171
172        The value returned from this method is passed to most other methods in this class. Use it to form a request e.g: set headers or query params.
173
174        :return: The token for the next page from the input response object. Returning None means there are no more pages to read in this response.
175        """
176
177    @abstractmethod
178    def path(
179        self,
180        *,
181        stream_state: Optional[Mapping[str, Any]] = None,
182        stream_slice: Optional[Mapping[str, Any]] = None,
183        next_page_token: Optional[Mapping[str, Any]] = None,
184    ) -> str:
185        """
186        Returns the URL path for the API endpoint e.g: if you wanted to hit https://myapi.com/v1/some_entity then this should return "some_entity"
187        """
188
189    def request_params(
190        self,
191        stream_state: Optional[Mapping[str, Any]],
192        stream_slice: Optional[Mapping[str, Any]] = None,
193        next_page_token: Optional[Mapping[str, Any]] = None,
194    ) -> MutableMapping[str, Any]:
195        """
196        Override this method to define the query parameters that should be set on an outgoing HTTP request given the inputs.
197
198        E.g: you might want to define query parameters for paging if next_page_token is not None.
199        """
200        return {}
201
202    def request_headers(
203        self,
204        stream_state: Optional[Mapping[str, Any]],
205        stream_slice: Optional[Mapping[str, Any]] = None,
206        next_page_token: Optional[Mapping[str, Any]] = None,
207    ) -> Mapping[str, Any]:
208        """
209        Override to return any non-auth headers. Authentication headers will overwrite any overlapping headers returned from this method.
210        """
211        return {}
212
213    def request_body_data(
214        self,
215        stream_state: Optional[Mapping[str, Any]],
216        stream_slice: Optional[Mapping[str, Any]] = None,
217        next_page_token: Optional[Mapping[str, Any]] = None,
218    ) -> Optional[Union[Mapping[str, Any], str]]:
219        """
220        Override when creating POST/PUT/PATCH requests to populate the body of the request with a non-JSON payload.
221
222        If returns a ready text that it will be sent as is.
223        If returns a dict that it will be converted to a urlencoded form.
224        E.g. {"key1": "value1", "key2": "value2"} => "key1=value1&key2=value2"
225
226        At the same time only one of the 'request_body_data' and 'request_body_json' functions can be overridden.
227        """
228        return None
229
230    def request_body_json(
231        self,
232        stream_state: Optional[Mapping[str, Any]],
233        stream_slice: Optional[Mapping[str, Any]] = None,
234        next_page_token: Optional[Mapping[str, Any]] = None,
235    ) -> Optional[Mapping[str, Any]]:
236        """
237        Override when creating POST/PUT/PATCH requests to populate the body of the request with a JSON payload.
238
239        At the same time only one of the 'request_body_data' and 'request_body_json' functions can be overridden.
240        """
241        return None
242
243    def request_kwargs(
244        self,
245        stream_state: Optional[Mapping[str, Any]],
246        stream_slice: Optional[Mapping[str, Any]] = None,
247        next_page_token: Optional[Mapping[str, Any]] = None,
248    ) -> Mapping[str, Any]:
249        """
250        Override to return a mapping of keyword arguments to be used when creating the HTTP request.
251        Any option listed in https://docs.python-requests.org/en/latest/api/#requests.adapters.BaseAdapter.send for can be returned from
252        this method. Note that these options do not conflict with request-level options such as headers, request params, etc..
253        """
254        return {}
255
256    @abstractmethod
257    def parse_response(
258        self,
259        response: requests.Response,
260        *,
261        stream_state: Mapping[str, Any],
262        stream_slice: Optional[Mapping[str, Any]] = None,
263        next_page_token: Optional[Mapping[str, Any]] = None,
264    ) -> Iterable[Mapping[str, Any]]:
265        """
266        Parses the raw response object into a list of records.
267        By default, this returns an iterable containing the input. Override to parse differently.
268        :param response:
269        :param stream_state:
270        :param stream_slice:
271        :param next_page_token:
272        :return: An iterable containing the parsed response
273        """
274
275    def get_backoff_strategy(self) -> Optional[Union[BackoffStrategy, List[BackoffStrategy]]]:
276        """
277        Used to initialize Adapter to avoid breaking changes.
278        If Stream has a `backoff_time` method implementation, we know this stream uses old (pre-HTTPClient) backoff handlers and thus an adapter is needed.
279
280        Override to provide custom BackoffStrategy
281        :return Optional[BackoffStrategy]:
282        """
283        if hasattr(self, "backoff_time"):
284            return HttpStreamAdapterBackoffStrategy(self)
285        else:
286            return None
287
288    def get_error_handler(self) -> Optional[ErrorHandler]:
289        """
290        Used to initialize Adapter to avoid breaking changes.
291        If Stream has a `should_retry` method implementation, we know this stream uses old (pre-HTTPClient) error handlers and thus an adapter is needed.
292
293        Override to provide custom ErrorHandler
294        :return Optional[ErrorHandler]:
295        """
296        if hasattr(self, "should_retry"):
297            error_handler = HttpStreamAdapterHttpStatusErrorHandler(
298                stream=self,
299                logger=logging.getLogger(),
300                max_retries=self.max_retries,
301                max_time=timedelta(seconds=self.max_time or 0),
302            )
303            return error_handler
304        else:
305            return None
306
307    @classmethod
308    def _join_url(cls, url_base: str, path: str) -> str:
309        return urljoin(url_base, path)
310
311    @classmethod
312    def parse_response_error_message(cls, response: requests.Response) -> Optional[str]:
313        """
314        Parses the raw response object from a failed request into a user-friendly error message.
315        By default, this method tries to grab the error message from JSON responses by following common API patterns. Override to parse differently.
316
317        :param response:
318        :return: A user-friendly message that indicates the cause of the error
319        """
320
321        # default logic to grab error from common fields
322        def _try_get_error(value: Optional[JsonType]) -> Optional[str]:
323            if isinstance(value, str):
324                return value
325            elif isinstance(value, list):
326                errors_in_value = [_try_get_error(v) for v in value]
327                return ", ".join(v for v in errors_in_value if v is not None)
328            elif isinstance(value, dict):
329                new_value = (
330                    value.get("message")
331                    or value.get("messages")
332                    or value.get("error")
333                    or value.get("errors")
334                    or value.get("failures")
335                    or value.get("failure")
336                    or value.get("detail")
337                )
338                return _try_get_error(new_value)
339            return None
340
341        try:
342            body = response.json()
343            return _try_get_error(body)
344        except requests.exceptions.JSONDecodeError:
345            return None
346
347    def get_error_display_message(self, exception: BaseException) -> Optional[str]:
348        """
349        Retrieves the user-friendly display message that corresponds to an exception.
350        This will be called when encountering an exception while reading records from the stream, and used to build the AirbyteTraceMessage.
351
352        The default implementation of this method only handles HTTPErrors by passing the response to self.parse_response_error_message().
353        The method should be overriden as needed to handle any additional exception types.
354
355        :param exception: The exception that was raised
356        :return: A user-friendly message that indicates the cause of the error
357        """
358        if isinstance(exception, requests.HTTPError) and exception.response is not None:
359            return self.parse_response_error_message(exception.response)
360        return None
361
362    def read_records(
363        self,
364        sync_mode: SyncMode,
365        cursor_field: Optional[List[str]] = None,
366        stream_slice: Optional[Mapping[str, Any]] = None,
367        stream_state: Optional[Mapping[str, Any]] = None,
368    ) -> Iterable[StreamData]:
369        # A cursor_field indicates this is an incremental stream which offers better checkpointing than RFR enabled via the cursor
370        if self.cursor_field or not isinstance(self.get_cursor(), ResumableFullRefreshCursor):
371            yield from self._read_pages(
372                lambda req, res, state, _slice: self.parse_response(
373                    res, stream_slice=_slice, stream_state=state
374                ),
375                stream_slice,
376                stream_state,
377            )
378        else:
379            yield from self._read_single_page(
380                lambda req, res, state, _slice: self.parse_response(
381                    res, stream_slice=_slice, stream_state=state
382                ),
383                stream_slice,
384                stream_state,
385            )
386
387    @property
388    def state(self) -> MutableMapping[str, Any]:
389        cursor = self.get_cursor()
390        if cursor:
391            return cursor.get_stream_state()  # type: ignore
392        return self._state
393
394    @state.setter
395    def state(self, value: MutableMapping[str, Any]) -> None:
396        cursor = self.get_cursor()
397        if cursor:
398            cursor.set_initial_state(value)
399        self._state = value
400
401    def get_cursor(self) -> Optional[Cursor]:
402        # I don't love that this is semi-stateful but not sure what else to do. We don't know exactly what type of cursor to
403        # instantiate when creating the class. We can make a few assumptions like if there is a cursor_field which implies
404        # incremental, but we don't know until runtime if this is a substream. Ideally, a stream should explicitly define
405        # its cursor, but because we're trying to automatically apply RFR we're stuck with this logic where we replace the
406        # cursor at runtime once we detect this is a substream based on self.has_multiple_slices being reassigned
407        if self.has_multiple_slices and isinstance(self.cursor, ResumableFullRefreshCursor):
408            self.cursor = SubstreamResumableFullRefreshCursor()
409            return self.cursor
410        else:
411            return self.cursor
412
413    def _read_pages(
414        self,
415        records_generator_fn: Callable[
416            [
417                requests.PreparedRequest,
418                requests.Response,
419                Mapping[str, Any],
420                Optional[Mapping[str, Any]],
421            ],
422            Iterable[StreamData],
423        ],
424        stream_slice: Optional[Mapping[str, Any]] = None,
425        stream_state: Optional[Mapping[str, Any]] = None,
426    ) -> Iterable[StreamData]:
427        stream_state = stream_state or {}
428        pagination_complete = False
429        next_page_token = None
430        while not pagination_complete:
431            request, response = self._fetch_next_page(stream_slice, stream_state, next_page_token)
432            yield from records_generator_fn(request, response, stream_state, stream_slice)
433
434            next_page_token = self.next_page_token(response)
435            if not next_page_token:
436                pagination_complete = True
437
438        cursor = self.get_cursor()
439        if cursor and isinstance(cursor, SubstreamResumableFullRefreshCursor):
440            partition, _, _ = self._extract_slice_fields(stream_slice=stream_slice)
441            # Substreams checkpoint state by marking an entire parent partition as completed so that on the subsequent attempt
442            # after a failure, completed parents are skipped and the sync can make progress
443            cursor.close_slice(StreamSlice(cursor_slice={}, partition=partition))
444
445        # Always return an empty generator just in case no records were ever yielded
446        yield from []
447
448    def _read_single_page(
449        self,
450        records_generator_fn: Callable[
451            [
452                requests.PreparedRequest,
453                requests.Response,
454                Mapping[str, Any],
455                Optional[Mapping[str, Any]],
456            ],
457            Iterable[StreamData],
458        ],
459        stream_slice: Optional[Mapping[str, Any]] = None,
460        stream_state: Optional[Mapping[str, Any]] = None,
461    ) -> Iterable[StreamData]:
462        partition, cursor_slice, remaining_slice = self._extract_slice_fields(
463            stream_slice=stream_slice
464        )
465        stream_state = stream_state or {}
466        next_page_token = cursor_slice or None
467
468        request, response = self._fetch_next_page(remaining_slice, stream_state, next_page_token)
469        yield from records_generator_fn(request, response, stream_state, remaining_slice)
470
471        next_page_token = self.next_page_token(response) or {
472            "__ab_full_refresh_sync_complete": True
473        }
474
475        cursor = self.get_cursor()
476        if cursor:
477            cursor.close_slice(StreamSlice(cursor_slice=next_page_token, partition=partition))
478
479        # Always return an empty generator just in case no records were ever yielded
480        yield from []
481
482    @staticmethod
483    def _extract_slice_fields(
484        stream_slice: Optional[Mapping[str, Any]],
485    ) -> tuple[Mapping[str, Any], Mapping[str, Any], Mapping[str, Any]]:
486        if not stream_slice:
487            return {}, {}, {}
488
489        if isinstance(stream_slice, StreamSlice):
490            partition = stream_slice.partition
491            cursor_slice = stream_slice.cursor_slice
492            remaining = {k: v for k, v in stream_slice.items()}
493        else:
494            # RFR streams that implement stream_slices() to generate stream slices in the legacy mapping format are converted into a
495            # structured stream slice mapping by the LegacyCursorBasedCheckpointReader. The structured mapping object has separate
496            # fields for the partition and cursor_slice value
497            partition = stream_slice.get("partition", {})
498            cursor_slice = stream_slice.get("cursor_slice", {})
499            remaining = {
500                key: val
501                for key, val in stream_slice.items()
502                if key != "partition" and key != "cursor_slice"
503            }
504        return partition, cursor_slice, remaining
505
506    def _fetch_next_page(
507        self,
508        stream_slice: Optional[Mapping[str, Any]] = None,
509        stream_state: Optional[Mapping[str, Any]] = None,
510        next_page_token: Optional[Mapping[str, Any]] = None,
511    ) -> Tuple[requests.PreparedRequest, requests.Response]:
512        request, response = self._http_client.send_request(
513            http_method=self.http_method,
514            url=self._join_url(
515                self.url_base,
516                self.path(
517                    stream_state=stream_state,
518                    stream_slice=stream_slice,
519                    next_page_token=next_page_token,
520                ),
521            ),
522            request_kwargs=self.request_kwargs(
523                stream_state=stream_state,
524                stream_slice=stream_slice,
525                next_page_token=next_page_token,
526            ),
527            headers=self.request_headers(
528                stream_state=stream_state,
529                stream_slice=stream_slice,
530                next_page_token=next_page_token,
531            ),
532            params=self.request_params(
533                stream_state=stream_state,
534                stream_slice=stream_slice,
535                next_page_token=next_page_token,
536            ),
537            json=self.request_body_json(
538                stream_state=stream_state,
539                stream_slice=stream_slice,
540                next_page_token=next_page_token,
541            ),
542            data=self.request_body_data(
543                stream_state=stream_state,
544                stream_slice=stream_slice,
545                next_page_token=next_page_token,
546            ),
547            dedupe_query_params=True,
548            log_formatter=self.get_log_formatter(),
549            exit_on_rate_limit=self.exit_on_rate_limit,
550        )
551
552        return request, response
553
554    def get_log_formatter(self) -> Optional[Callable[[requests.Response], Any]]:
555        """
556
557        :return Optional[Callable[[requests.Response], Any]]: Function that will be used in logging inside HttpClient
558        """
559        return None