airbyte_cdk.sources.declarative.retrievers.simple_retriever

  1#
  2# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  3#
  4
  5import datetime
  6import json
  7import logging
  8from collections import defaultdict
  9from dataclasses import InitVar, dataclass, field
 10from functools import partial
 11from typing import (
 12    Any,
 13    Callable,
 14    Iterable,
 15    List,
 16    Mapping,
 17    MutableMapping,
 18    Optional,
 19    Set,
 20    Tuple,
 21    Union,
 22)
 23
 24import requests
 25from typing_extensions import deprecated
 26
 27from airbyte_cdk.models import FailureType
 28from airbyte_cdk.sources.declarative.extractors.http_selector import HttpSelector
 29from airbyte_cdk.sources.declarative.extractors.record_filter import (
 30    ClientSideIncrementalRecordFilterDecorator,
 31)
 32from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 33from airbyte_cdk.sources.declarative.partition_routers.single_partition_router import (
 34    SinglePartitionRouter,
 35)
 36from airbyte_cdk.sources.declarative.requesters.paginators.no_pagination import NoPagination
 37from airbyte_cdk.sources.declarative.requesters.paginators.paginator import (
 38    Paginator,
 39    page_size_override_kwargs,
 40    stream_slice_kwargs,
 41)
 42from airbyte_cdk.sources.declarative.requesters.query_properties import QueryProperties
 43from airbyte_cdk.sources.declarative.requesters.request_options import (
 44    DefaultRequestOptionsProvider,
 45    RequestOptionsProvider,
 46)
 47from airbyte_cdk.sources.declarative.requesters.requester import Requester
 48from airbyte_cdk.sources.declarative.retrievers.page_size_reducer import (
 49    PageSizeReducer,
 50    PageSizeReduction,
 51)
 52from airbyte_cdk.sources.declarative.retrievers.pagination_tracker import PaginationTracker
 53from airbyte_cdk.sources.declarative.retrievers.request_window_splitting import (
 54    RequestWindowSplitting,
 55)
 56from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
 57from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer
 58from airbyte_cdk.sources.source import ExperimentalClassWarning
 59from airbyte_cdk.sources.streams.core import StreamData
 60from airbyte_cdk.sources.streams.http.page_size_reduction_exception import (
 61    PageSizeReductionNotSupportedException,
 62    PageSizeReductionRequiredException,
 63)
 64from airbyte_cdk.sources.streams.http.pagination_reset_exception import (
 65    PaginationResetRequiredException,
 66)
 67from airbyte_cdk.sources.streams.http.request_window_split_exception import (
 68    RequestWindowSplitNotSupportedException,
 69    RequestWindowSplitRequiredException,
 70)
 71from airbyte_cdk.sources.types import Config, Record, StreamSlice
 72from airbyte_cdk.utils.mapping_helpers import combine_mappings
 73from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 74
 75FULL_REFRESH_SYNC_COMPLETE_KEY = "__ab_full_refresh_sync_complete"
 76LOGGER = logging.getLogger("airbyte")
 77
 78# A defense-in-depth bound independent of any `request_window_splitter`'s own no-progress guard: protects
 79# against a misbehaving custom cursor whose children don't actually shrink the window. Not user-configurable.
 80_MAX_REQUEST_WINDOW_SPLIT_DEPTH = 10
 81
 82
 83@dataclass(frozen=True)
 84class PageCompleted:
 85    """Describe a completed response page observed by an incremental cursor."""
 86
 87    stream_slice: StreamSlice
 88    response: requests.Response
 89    pagination_exhausted: bool
 90
 91
 92@dataclass
 93class SimpleRetriever(Retriever):
 94    """
 95    Retrieves records by synchronously sending requests to fetch records.
 96
 97    The retriever acts as an orchestrator between the requester, the record selector, the paginator, and the stream slicer.
 98
 99    For each stream slice, submit requests until there are no more pages of records to fetch.
100
101    This retriever currently inherits from HttpStream to reuse the request submission and pagination machinery.
102    As a result, some of the parameters passed to some methods are unused.
103    The two will be decoupled in a future release.
104
105    Attributes:
106        stream_name (str): The stream's name
107        stream_primary_key (Optional[Union[str, List[str], List[List[str]]]]): The stream's primary key
108        requester (Requester): The HTTP requester
109        record_selector (HttpSelector): The record selector
110        paginator (Optional[Paginator]): The paginator
111        stream_slicer (Optional[StreamSlicer]): The stream slicer
112        parameters (Mapping[str, Any]): Additional runtime parameters to be used for string interpolation
113        post_pagination_filter (Optional[ClientSideIncrementalRecordFilterDecorator]): Set for data feed streams only.
114            Records the cursor considers already synced are dropped once pagination has observed them
115        page_size_reduction (Optional[PageSizeReduction]): How much to shrink the page size when an error handler
116            resolves to `ResponseAction.REDUCE_PAGE_SIZE`. `None` disables page size reduction entirely.
117            It is immutable configuration; the page size in effect lives in a `PageSizeReducer` that
118            `_read_pages` creates per call, so the retriever and its paginator - both shared by every
119            partition of the stream, read concurrently - stay stateless. When `page_size_reduction` is
120            `None` no reducer is created and `_read_pages` keeps its previous behaviour
121        request_window_splitting (Optional[RequestWindowSplitting]): Policy applied when an error handler
122            resolves to `ResponseAction.SPLIT_REQUEST_WINDOW`, or when `RequestWindowSplitRequiredException`
123            is raised directly by custom code. `None` disables request-window splitting entirely: `read_records`
124            converts the exception into `RequestWindowSplitNotSupportedException` instead of attempting a
125            split it cannot perform.
126        request_window_splitter: Bound method - normally the stream's own cursor's `split_request_window` -
127            asked to replace a failing `StreamSlice` with smaller children. `None` has the same effect as
128            `request_window_splitting` being `None`.
129        page_observer (Optional[Callable[[PageCompleted], None]]): Observes completed response pages; set
130            only on streams using `SyncTokenCursor` and `None` otherwise.
131    """
132
133    requester: Requester
134    record_selector: HttpSelector
135    config: Config
136    parameters: InitVar[Mapping[str, Any]]
137    name: str
138    _name: Union[InterpolatedString, str] = field(init=False, repr=False, default="")
139    primary_key: Optional[Union[str, List[str], List[List[str]]]]
140    _primary_key: str = field(init=False, repr=False, default="")
141    paginator: Optional[Paginator] = None
142    stream_slicer: StreamSlicer = field(
143        default_factory=lambda: SinglePartitionRouter(parameters={})
144    )
145    request_option_provider: RequestOptionsProvider = field(
146        default_factory=lambda: DefaultRequestOptionsProvider(parameters={})
147    )
148    ignore_stream_slicer_parameters_on_paginated_requests: bool = False
149    additional_query_properties: Optional[QueryProperties] = None
150    log_formatter: Optional[Callable[[requests.Response], Any]] = None
151    pagination_tracker_factory: Callable[[], PaginationTracker] = field(
152        default_factory=lambda: lambda: PaginationTracker()
153    )
154    post_pagination_filter: Optional[ClientSideIncrementalRecordFilterDecorator] = None
155    page_size_reduction: Optional[PageSizeReduction] = None
156    request_window_splitting: Optional[RequestWindowSplitting] = None
157    request_window_splitter: Optional[
158        Callable[[StreamSlice, Optional[datetime.timedelta]], Optional[List[StreamSlice]]]
159    ] = None
160    page_observer: Optional[Callable[["PageCompleted"], None]] = None
161
162    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
163        self._paginator = self.paginator or NoPagination(parameters=parameters)
164        self._parameters = parameters
165        self._name = (
166            InterpolatedString(self._name, parameters=parameters)
167            if isinstance(self._name, str)
168            else self._name
169        )
170
171    @property  # type: ignore
172    def name(self) -> str:
173        """
174        :return: Stream name
175        """
176        return (
177            str(self._name.eval(self.config))
178            if isinstance(self._name, InterpolatedString)
179            else self._name
180        )
181
182    @name.setter
183    def name(self, value: str) -> None:
184        if not isinstance(value, property):
185            self._name = value
186
187    def _get_mapping(
188        self, method: Callable[..., Optional[Union[Mapping[str, Any], str]]], **kwargs: Any
189    ) -> Tuple[Union[Mapping[str, Any], str], Set[str]]:
190        """
191        Get mapping from the provided method, and get the keys of the mapping.
192        If the method returns a string, it will return the string and an empty set.
193        If the method returns a dict, it will return the dict and its keys.
194        """
195        mapping = method(**kwargs) or {}
196        keys = set(mapping.keys()) if not isinstance(mapping, str) else set()
197        return mapping, keys
198
199    def _get_request_options(
200        self,
201        stream_slice: Optional[StreamSlice],
202        next_page_token: Optional[Mapping[str, Any]],
203        paginator_method: Callable[..., Optional[Union[Mapping[str, Any], str]]],
204        stream_slicer_method: Callable[..., Optional[Union[Mapping[str, Any], str]]],
205        page_size_override: Optional[int] = None,
206    ) -> Union[Mapping[str, Any], str]:
207        """
208        Get the request_option from the paginator and the stream slicer.
209        Raise a ValueError if there's a key collision
210        Returned merged mapping otherwise
211        """
212        is_body_json = paginator_method.__name__ == "get_request_body_json"
213
214        mappings = [
215            paginator_method(
216                stream_slice=stream_slice,
217                next_page_token=next_page_token,
218                **page_size_override_kwargs(page_size_override),
219            ),
220        ]
221        if not next_page_token or not self.ignore_stream_slicer_parameters_on_paginated_requests:
222            mappings.append(
223                stream_slicer_method(
224                    stream_slice=stream_slice,
225                    next_page_token=next_page_token,
226                )
227            )
228        return combine_mappings(mappings, allow_same_value_merge=is_body_json)
229
230    def _request_headers(
231        self,
232        stream_slice: Optional[StreamSlice] = None,
233        next_page_token: Optional[Mapping[str, Any]] = None,
234        page_size_override: Optional[int] = None,
235    ) -> Mapping[str, Any]:
236        """
237        Specifies request headers.
238        Authentication headers will overwrite any overlapping headers returned from this method.
239        """
240        headers = self._get_request_options(
241            stream_slice,
242            next_page_token,
243            self._paginator.get_request_headers,
244            self.request_option_provider.get_request_headers,
245            **page_size_override_kwargs(page_size_override),
246        )
247        if isinstance(headers, str):
248            raise ValueError("Request headers cannot be a string")
249        return {str(k): str(v) for k, v in headers.items()}
250
251    def _request_params(
252        self,
253        stream_slice: Optional[StreamSlice] = None,
254        next_page_token: Optional[Mapping[str, Any]] = None,
255        page_size_override: Optional[int] = None,
256    ) -> Mapping[str, Any]:
257        """
258        Specifies the query parameters that should be set on an outgoing HTTP request given the inputs.
259
260        E.g: you might want to define query parameters for paging if next_page_token is not None.
261        """
262        params = self._get_request_options(
263            stream_slice,
264            next_page_token,
265            self._paginator.get_request_params,
266            self.request_option_provider.get_request_params,
267            **page_size_override_kwargs(page_size_override),
268        )
269        if isinstance(params, str):
270            raise ValueError("Request params cannot be a string")
271        return params
272
273    def _request_body_data(
274        self,
275        stream_slice: Optional[StreamSlice] = None,
276        next_page_token: Optional[Mapping[str, Any]] = None,
277        page_size_override: Optional[int] = None,
278    ) -> Union[Mapping[str, Any], str]:
279        """
280        Specifies how to populate the body of the request with a non-JSON payload.
281
282        If returns a ready text that it will be sent as is.
283        If returns a dict that it will be converted to a urlencoded form.
284        E.g. {"key1": "value1", "key2": "value2"} => "key1=value1&key2=value2"
285
286        At the same time only one of the 'request_body_data' and 'request_body_json' functions can be overridden.
287        """
288        return self._get_request_options(
289            stream_slice,
290            next_page_token,
291            self._paginator.get_request_body_data,
292            self.request_option_provider.get_request_body_data,
293            **page_size_override_kwargs(page_size_override),
294        )
295
296    def _request_body_json(
297        self,
298        stream_slice: Optional[StreamSlice] = None,
299        next_page_token: Optional[Mapping[str, Any]] = None,
300        page_size_override: Optional[int] = None,
301    ) -> Optional[Mapping[str, Any]]:
302        """
303        Specifies how to populate the body of the request with a JSON payload.
304
305        At the same time only one of the 'request_body_data' and 'request_body_json' functions can be overridden.
306        """
307        body_json = self._get_request_options(
308            stream_slice,
309            next_page_token,
310            self._paginator.get_request_body_json,
311            self.request_option_provider.get_request_body_json,
312            **page_size_override_kwargs(page_size_override),
313        )
314        if isinstance(body_json, str):
315            raise ValueError("Request body json cannot be a string")
316        return body_json
317
318    def _paginator_path(
319        self,
320        next_page_token: Optional[Mapping[str, Any]] = None,
321        stream_slice: Optional[StreamSlice] = None,
322    ) -> Optional[str]:
323        """
324        If the paginator points to a path, follow it, else return nothing so the requester is used.
325        :param next_page_token:
326        :return:
327        """
328        return self._paginator.path(
329            next_page_token=next_page_token,
330            stream_state={},  # stream_state as an interpolation context is deprecated
331            stream_slice=stream_slice,
332        )
333
334    def _parse_response(
335        self,
336        response: Optional[requests.Response],
337        records_schema: Mapping[str, Any],
338        stream_slice: Optional[StreamSlice] = None,
339        next_page_token: Optional[Mapping[str, Any]] = None,
340    ) -> Iterable[Record]:
341        if not response:
342            yield from []
343        else:
344            yield from self.record_selector.select_records(
345                response=response,
346                stream_state={},  # stream_state as an interpolation context is deprecated
347                records_schema=records_schema,
348                stream_slice=stream_slice,
349                next_page_token=next_page_token,
350            )
351
352    @property  # type: ignore
353    def primary_key(self) -> Optional[Union[str, List[str], List[List[str]]]]:
354        """The stream's primary key"""
355        return self._primary_key
356
357    @primary_key.setter
358    def primary_key(self, value: str) -> None:
359        if not isinstance(value, property):
360            self._primary_key = value
361
362    def _next_page_token(
363        self,
364        response: requests.Response,
365        last_page_size: int,
366        last_record: Optional[Record],
367        last_page_token_value: Optional[Any],
368        page_size_override: Optional[int] = None,
369        stream_slice: Optional[StreamSlice] = None,
370    ) -> Optional[Mapping[str, Any]]:
371        """
372        Specifies a pagination strategy.
373
374        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.
375
376        :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.
377        """
378        return self._paginator.next_page_token(
379            response=response,
380            last_page_size=last_page_size,
381            last_record=last_record,
382            last_page_token_value=last_page_token_value,
383            **page_size_override_kwargs(page_size_override),
384            **stream_slice_kwargs(self._paginator.next_page_token, stream_slice),
385        )
386
387    def _fetch_next_page(
388        self,
389        stream_slice: StreamSlice,
390        next_page_token: Optional[Mapping[str, Any]] = None,
391        page_size_override: Optional[int] = None,
392    ) -> Optional[requests.Response]:
393        return self.requester.send_request(
394            path=self._paginator_path(
395                next_page_token=next_page_token,
396                stream_slice=stream_slice,
397            ),
398            stream_state={},  # stream_state as an interpolation context is deprecated
399            stream_slice=stream_slice,
400            next_page_token=next_page_token,
401            request_headers=self._request_headers(
402                stream_slice=stream_slice,
403                next_page_token=next_page_token,
404                **page_size_override_kwargs(page_size_override),
405            ),
406            request_params=self._request_params(
407                stream_slice=stream_slice,
408                next_page_token=next_page_token,
409                **page_size_override_kwargs(page_size_override),
410            ),
411            request_body_data=self._request_body_data(
412                stream_slice=stream_slice,
413                next_page_token=next_page_token,
414                **page_size_override_kwargs(page_size_override),
415            ),
416            request_body_json=self._request_body_json(
417                stream_slice=stream_slice,
418                next_page_token=next_page_token,
419                **page_size_override_kwargs(page_size_override),
420            ),
421            log_formatter=self.log_formatter,
422        )
423
424    # This logic is similar to _read_pages in the HttpStream class. When making changes here, consider making changes there as well.
425    def _read_pages(
426        self,
427        records_generator_fn: Callable[[Optional[requests.Response]], Iterable[Record]],
428        stream_slice: StreamSlice,
429    ) -> Iterable[Record]:
430        original_stream_slice = stream_slice
431        pagination_tracker = self.pagination_tracker_factory()
432        page_size_reducer = (
433            PageSizeReducer(
434                self.page_size_reduction,
435                self._paginator.get_page_size(),
436                stream_name=self.name,
437            )
438            if self.page_size_reduction
439            else None
440        )
441        reset_pagination = False
442        reduce_page_size = False
443        next_page_token = self._get_initial_next_page_token()
444        while True:
445            page_size_override = page_size_reducer.page_size_override if page_size_reducer else None
446            merged_records: MutableMapping[str, Any] = defaultdict(dict)
447            last_page_size = 0
448            last_record: Optional[Record] = None
449
450            response = None
451            try:
452                if self.additional_query_properties:
453                    for (
454                        properties
455                    ) in self.additional_query_properties.get_request_property_chunks():
456                        stream_slice = StreamSlice(
457                            partition=stream_slice.partition or {},
458                            cursor_slice=stream_slice.cursor_slice or {},
459                            extra_fields={"query_properties": properties},
460                        )
461                        response = self._fetch_next_page(
462                            stream_slice,
463                            next_page_token,
464                            **page_size_override_kwargs(page_size_override),
465                        )
466
467                        for current_record in records_generator_fn(response):
468                            if self.additional_query_properties.property_chunking:
469                                merge_key = self.additional_query_properties.property_chunking.get_merge_key(
470                                    current_record
471                                )
472                                if merge_key:
473                                    _deep_merge(merged_records[merge_key], current_record)
474                                else:
475                                    # We should still emit records even if the record did not have a merge key
476                                    pagination_tracker.observe(current_record)
477                                    last_page_size += 1
478                                    last_record = current_record
479                                    yield current_record
480                            else:
481                                pagination_tracker.observe(current_record)
482                                last_page_size += 1
483                                last_record = current_record
484                                yield current_record
485
486                    for merged_record in merged_records.values():
487                        record = Record(
488                            data=merged_record, stream_name=self.name, associated_slice=stream_slice
489                        )
490                        pagination_tracker.observe(record)
491                        last_page_size += 1
492                        last_record = record
493                        yield record
494                else:
495                    response = self._fetch_next_page(
496                        stream_slice,
497                        next_page_token,
498                        **page_size_override_kwargs(page_size_override),
499                    )
500                    for current_record in records_generator_fn(response):
501                        pagination_tracker.observe(current_record)
502                        last_page_size += 1
503                        last_record = current_record
504                        yield current_record
505            except PaginationResetRequiredException:
506                reset_pagination = True
507            except PageSizeReductionRequiredException as exception:
508                if page_size_reducer is None:
509                    # The action can be attached to a requester we cannot validate at config time, such as one
510                    # built by a custom error handler or a CustomRequester. Re-raise as the misconfiguration it
511                    # is: the exception being handled is the neutral "the API asked for a smaller page" signal.
512                    raise PageSizeReductionNotSupportedException(stream_name=self.name)
513                if last_page_size:
514                    # The reduction is safe only because it re-issues a page whose records were not emitted.
515                    # Every in-CDK way of reaching this raises from `_fetch_next_page`, before the record loop,
516                    # and the factory rejects the manifest constructs that would not, but a custom extractor,
517                    # filter or transformation can issue its own request from inside the record generator.
518                    # Re-issuing the page then duplicates the records already yielded, so fail instead.
519                    raise AirbyteTracedException(
520                        internal_message=f"Stream {self.name} requested a page size reduction after {last_page_size} records of the page had already been emitted",
521                        message=f"Stream {self.name} asked for a smaller page size in the middle of a page. The page cannot be requested again without duplicating the records already read from it. Move the REDUCE_PAGE_SIZE action to the error handler of the stream's main requester.",
522                        failure_type=FailureType.config_error,
523                    )
524                # Raises once the page size cannot be reduced any further and the retries allowed at that
525                # floor are spent, which is what stops the loop when the API keeps failing.
526                page_size_reducer.reduce(exception.error_message)
527                reduce_page_size = True
528            else:
529                if page_size_reducer:
530                    page_size_reducer.on_successful_page()
531                if not response:
532                    break
533
534            if reduce_page_size:
535                # Retry the very same page: neither the token nor the slice change, only the page size does -
536                # and not even that once the reducer is at its floor and only waiting is left.
537                reduce_page_size = False
538                continue
539
540            if reset_pagination or pagination_tracker.has_reached_limit():
541                next_page_token = self._get_initial_next_page_token()
542                previous_slice = stream_slice
543                stream_slice = pagination_tracker.reduce_slice_range_if_possible(
544                    stream_slice, original_stream_slice
545                )
546                LOGGER.info(
547                    f"Hitting PaginationReset event. StreamSlice used will go from {previous_slice} to {stream_slice}"
548                )
549                reset_pagination = False
550            else:
551                last_page_token_value = (
552                    next_page_token.get("next_page_token") if next_page_token else None
553                )
554                next_page_token = self._next_page_token(
555                    response=response,  # type:ignore # we are breaking from the loop on the try/else if there are no response so this should be fine
556                    last_page_size=last_page_size,
557                    last_record=last_record,
558                    last_page_token_value=last_page_token_value,
559                    **page_size_override_kwargs(page_size_override),
560                    **stream_slice_kwargs(self._next_page_token, stream_slice),
561                )
562                if response is not None and self.page_observer:
563                    self.page_observer(
564                        PageCompleted(
565                            stream_slice=stream_slice,
566                            response=response,
567                            pagination_exhausted=not next_page_token,
568                        )
569                    )
570                if not next_page_token:
571                    break
572
573        # Always return an empty generator just in case no records were ever yielded
574        yield from []
575
576    def _get_initial_next_page_token(self) -> Optional[Mapping[str, Any]]:
577        initial_token = self._paginator.get_initial_token()
578        next_page_token = {"next_page_token": initial_token} if initial_token is not None else None
579        return next_page_token
580
581    def read_records(
582        self,
583        records_schema: Mapping[str, Any],
584        stream_slice: Optional[StreamSlice] = None,
585    ) -> Iterable[StreamData]:
586        """
587        Fetch a stream's records from an HTTP API source
588
589        :param records_schema: json schema to describe record
590        :param stream_slice: The stream slice to read data for
591        :return: The records read from the API source
592        """
593        _slice = stream_slice or StreamSlice(partition={}, cursor_slice={})  # None-check
594        yield from self._read_records_or_split_request_window(
595            records_schema, _slice, _slice, depth=0
596        )
597
598    def _read_records_or_split_request_window(
599        self,
600        records_schema: Mapping[str, Any],
601        stream_slice: StreamSlice,
602        original_slice: StreamSlice,
603        depth: int,
604    ) -> Iterable[StreamData]:
605        """
606        Read `stream_slice` to completion, replacing it with smaller children and recursing into each of them in
607        turn when a `RequestWindowSplitRequiredException` is raised while reading it.
608
609        Each recursive call re-enters this method - and, through it, `_read_pages` - from scratch for the child
610        slice it is given, so every child gets its own paginator token, `PaginationTracker`, and
611        `PageSizeReducer` for free: nothing here needs to reset that state explicitly. Because the whole
612        recursion lives inside the single generator `DeclarativePartition.read()` consumes,
613        `PartitionReader.process_partition()` only calls `cursor.close_partition()` once, after this generator is
614        fully exhausted - so a failure anywhere in the recursion (a child that cannot be read, or a window that
615        cannot be split any further) propagates out without ever checkpointing the original partition.
616
617        `original_slice` is threaded through the recursion unchanged so every record yielded, however deep the
618        recursion went to produce it, is re-stamped with the partition's own slice rather than the child slice it
619        was actually read against - see `_reassociate_with_original_slice`. `depth` bounds the recursion
620        independently of `request_window_splitter`'s own no-progress guard: it is enforced here so a
621        misbehaving custom cursor cannot recurse indefinitely regardless of what that guard does or does not
622        catch.
623        """
624        record_generator = partial(
625            self._parse_records,
626            stream_slice=stream_slice,
627            records_schema=records_schema,
628        )
629
630        emitted_count = 0
631        try:
632            records: Iterable[Mapping[str, Any]] = self._read_pages(record_generator, stream_slice)
633            if self.post_pagination_filter:
634                # A data feed paginates until it reaches a record older than the cursor, so the page that triggers the stop
635                # condition still holds already-synced records. Those are filtered here rather than in the record selector
636                # so that the paginator keeps seeing the whole page: the stop condition is evaluated on the last record of
637                # the page, which is precisely one of the records being dropped. Two consequences of filtering this late:
638                # the pagination tracker observes the dropped records, and a `file_uploader` on the record selector has
639                # already uploaded their files by the time they are dropped.
640                records = self.post_pagination_filter.filter_records(
641                    records,
642                    # the filter is only used for its cursor comparison, which does not read the stream state
643                    stream_state={},
644                    stream_slice=stream_slice,
645                )
646            for record in records:
647                emitted_count += 1
648                yield self._reassociate_with_original_slice(record, original_slice)
649        except RequestWindowSplitRequiredException as exception:
650            if self.request_window_splitting is None or self.request_window_splitter is None:
651                raise RequestWindowSplitNotSupportedException(stream_name=self.name) from exception
652
653            if emitted_count:
654                # Splitting and re-reading the window re-emits these records: a failed partition is never
655                # checkpointed, so the next attempt would re-emit them anyway. Always replaying makes forward
656                # progress instead of retrying the same oversized window forever; primary-key dedup downstream
657                # handles the duplicates either way.
658                LOGGER.warning(
659                    f"Stream {self.name} already emitted {emitted_count} record(s) from {stream_slice} before "
660                    f"it was rejected; splitting and re-reading the window may re-emit them."
661                )
662
663            rejection = f": {exception.error_message}" if exception.error_message else ""
664            if depth >= _MAX_REQUEST_WINDOW_SPLIT_DEPTH:
665                # Logged separately from the exception below so it stays visible even if the trace message's
666                # internal_message isn't surfaced by whatever catches it.
667                LOGGER.warning(
668                    f"Stream {self.name} hit the maximum request window split depth "
669                    f"({_MAX_REQUEST_WINDOW_SPLIT_DEPTH}) while splitting {original_slice}."
670                )
671                raise AirbyteTracedException(
672                    internal_message=f"Stream {self.name} exceeded the maximum request window split depth of {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} while splitting {original_slice}; the API still rejected {stream_slice}{rejection}",
673                    # `transient_error`, so the only remediation is the connector's own, if it defined one.
674                    message=self._with_request_window_failure_message(
675                        f"Stream {self.name} could not split its request window to a size the API accepts "
676                        f"within {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} splits."
677                    ),
678                    failure_type=FailureType.transient_error,
679                ) from exception
680
681            children = self.request_window_splitter(
682                stream_slice, self.request_window_splitting.min_split_window
683            )
684            if children is None:
685                min_split_window_note = (
686                    f", its configured `min_split_window` ({self.request_window_splitting.min_split_window})"
687                    if self.request_window_splitting.min_split_window
688                    else ""
689                )
690                raise AirbyteTracedException(
691                    internal_message=f"Stream {self.name} could not split its request window {stream_slice} any further; the API rejected it{rejection}",
692                    # `transient_error` regardless of how the triggering response was classified: exhaustion is
693                    # never the user's fault, matching `PageSizeReducer`'s equivalent exhaustion branch.
694                    message=self._with_request_window_failure_message(
695                        f"The API kept rejecting stream {self.name}'s request window even at the smallest window "
696                        f"its cursor allows{min_split_window_note}, or splitting the window further would not "
697                        f"make progress."
698                    ),
699                    failure_type=FailureType.transient_error,
700                ) from exception
701
702            LOGGER.info(
703                f"Stream {self.name}: the API rejected request window {stream_slice} (split depth {depth}); "
704                f"reducing it to {children} and reading each in turn."
705            )
706            for child in children:
707                yield from self._read_records_or_split_request_window(
708                    records_schema, child, original_slice, depth + 1
709                )
710
711    def _with_request_window_failure_message(self, message: str) -> str:
712        """
713        :return: the message followed by `request_window_splitting`'s configured `failure_message`, when it
714            defined one - mirroring `PageSizeReducer._with_failure_message`
715        """
716        assert self.request_window_splitting is not None
717        failure_message = (self.request_window_splitting.failure_message or "").strip()
718        if not failure_message:
719            return message
720        return f"{message} {failure_message}"
721
722    @staticmethod
723    def _reassociate_with_original_slice(
724        record: StreamData, original_slice: StreamSlice
725    ) -> StreamData:
726        """
727        Records read while recursing into a split child window are parsed against that child's `StreamSlice`,
728        so `RecordSelector` stamps them with the child as `associated_slice`. `DeclarativePartition.read()`
729        passes an already-built `Record` through unchanged rather than re-wrapping it, so left uncorrected the
730        child slice - not the partition's own slice - is what `ConcurrentCursor.observe()` would key its
731        per-partition bookkeeping by. `close_partition()` always looks that bookkeeping up by the partition's own
732        slice, so it would never find it, silently losing the precise most-recently-observed cursor value for a
733        split partition (recoverable in practice today only because callers of that lookup fall back to the
734        slice's end boundary when it is missing). Re-stamping every record with `original_slice` here restores
735        the same tracking a non-split read already gets, rather than relying on that fallback.
736        """
737        if isinstance(record, Record) and record.associated_slice is not original_slice:
738            return Record(
739                data=record.data,
740                stream_name=record.stream_name,
741                associated_slice=original_slice,
742                file_reference=record.file_reference,
743            )
744        return record
745
746    def _parse_records(
747        self,
748        response: Optional[requests.Response],
749        records_schema: Mapping[str, Any],
750        stream_slice: Optional[StreamSlice],
751    ) -> Iterable[Record]:
752        yield from self._parse_response(
753            response,
754            stream_slice=stream_slice,
755            records_schema=records_schema,
756        )
757
758    def must_deduplicate_query_params(self) -> bool:
759        return True
760
761    @staticmethod
762    def _to_partition_key(to_serialize: Any) -> str:
763        # separators have changed in Python 3.4. To avoid being impacted by further change, we explicitly specify our own value
764        return json.dumps(to_serialize, indent=None, separators=(",", ":"), sort_keys=True)
765
766
767def _deep_merge(
768    target: MutableMapping[str, Any], source: Union[Record, MutableMapping[str, Any]]
769) -> None:
770    """
771    Recursively merge two dictionaries, combining nested dictionaries instead of overwriting them.
772
773    :param target: The dictionary to merge into (modified in place)
774    :param source: The dictionary to merge from
775    """
776    for key, value in source.items():
777        if (
778            key in target
779            and isinstance(target[key], MutableMapping)
780            and isinstance(value, MutableMapping)
781        ):
782            _deep_merge(target[key], value)
783        else:
784            target[key] = value
785
786
787@deprecated(
788    "This class is experimental. Use at your own risk.",
789    category=ExperimentalClassWarning,
790)
791@dataclass
792class LazySimpleRetriever(SimpleRetriever):
793    """
794    A retriever that supports lazy loading from parent streams.
795    """
796
797    def _read_pages(
798        self,
799        records_generator_fn: Callable[[Optional[requests.Response]], Iterable[Record]],
800        stream_slice: StreamSlice,
801    ) -> Iterable[Record]:
802        response = stream_slice.extra_fields["child_response"]
803        if response:
804            last_page_size, last_record = 0, None
805            for record in records_generator_fn(response):  # type: ignore[call-arg] # only _parse_records expected as a func
806                last_page_size += 1
807                last_record = record
808                yield record
809
810            next_page_token = self._next_page_token(
811                response,
812                last_page_size,
813                last_record,
814                None,
815                **stream_slice_kwargs(self._next_page_token, stream_slice),
816            )
817            if next_page_token:
818                yield from self._paginate(
819                    next_page_token,
820                    records_generator_fn,
821                    stream_slice,
822                )
823
824            yield from []
825        else:
826            # coderabbit detected an interesting bug/gap where if we were to not get a child_response, we
827            # might recurse forever. This might not be the case, but it is worth noting that this code path
828            # isn't comprehensively tested.
829            yield from self._read_pages(records_generator_fn, stream_slice)
830
831    def _paginate(
832        self,
833        next_page_token: Any,
834        records_generator_fn: Callable[[Optional[requests.Response]], Iterable[Record]],
835        stream_slice: StreamSlice,
836    ) -> Iterable[Record]:
837        """Handle pagination by fetching subsequent pages."""
838        pagination_complete = False
839
840        while not pagination_complete:
841            response = self._fetch_next_page(stream_slice, next_page_token)
842            last_page_size, last_record = 0, None
843
844            for record in records_generator_fn(response):  # type: ignore[call-arg] # only _parse_records expected as a func
845                last_page_size += 1
846                last_record = record
847                yield record
848
849            if not response:
850                pagination_complete = True
851            else:
852                last_page_token_value = (
853                    next_page_token.get("next_page_token") if next_page_token else None
854                )
855                next_page_token = self._next_page_token(
856                    response,
857                    last_page_size,
858                    last_record,
859                    last_page_token_value,
860                    **stream_slice_kwargs(self._next_page_token, stream_slice),
861                )
862
863                if not next_page_token:
864                    pagination_complete = True