28@dataclass
 29class CursorPaginationStrategy(PaginationStrategy):
 30    """
 31    Pagination strategy that evaluates an interpolated string to define the next page token.
 32
 33    Attributes:
 34        page_size (Optional[Union[str, int]]): the number of records to request
 35        cursor_value (Union[InterpolatedString, str]): template string evaluating to the cursor value
 36        config (Config): connection config
 37        stop_condition (Optional[InterpolatedBoolean]): template string evaluating when to stop paginating
 38        decoder (Decoder): decoder to decode the response
 39    """
 40
 41    cursor_value: Union[InterpolatedString, str]
 42    config: Config
 43    parameters: InitVar[Mapping[str, Any]]
 44    page_size: Optional[Union[str, int]] = None
 45    stop_condition: Optional[Union[InterpolatedBoolean, str]] = None
 46    decoder: Decoder = field(
 47        default_factory=lambda: PaginationDecoderDecorator(decoder=JsonDecoder(parameters={}))
 48    )
 49
 50    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 51        if isinstance(self.cursor_value, str):
 52            self._cursor_value = InterpolatedString.create(self.cursor_value, parameters=parameters)
 53        else:
 54            self._cursor_value = self.cursor_value
 55        if isinstance(self.stop_condition, str):
 56            self._stop_condition: Optional[InterpolatedBoolean] = InterpolatedBoolean(
 57                condition=self.stop_condition, parameters=parameters
 58            )
 59        else:
 60            self._stop_condition = self.stop_condition
 61
 62        if isinstance(self.page_size, int) or (self.page_size is None):
 63            self._page_size = self.page_size
 64        else:
 65            page_size = InterpolatedString(self.page_size, parameters=parameters).eval(self.config)
 66            if not isinstance(page_size, int):
 67                raise Exception(f"{page_size} is of type {type(page_size)}. Expected {int}")
 68            self._page_size = page_size
 69
 70        if self._page_size is None:
 71            self._reject_unbound_page_size_variable()
 72
 73    def _reject_unbound_page_size_variable(self) -> None:
 74        """
 75        Fail construction when an expression reads `page_size` while the strategy declares none.
 76
 77        The variable holds the page size that was actually requested, so it is the one safe thing to compare
 78        `last_page_size` against while a `page_size_reduction` is shrinking the page. Without a declared
 79        `page_size` there is nothing to bind it to, and an unbound comparison does not raise: Jinja fails, the
 80        interpolation falls back to the raw template string, and a non-empty string is truthy. A
 81        `stop_condition` written that way stops after the first page and drops the rest of the partition
 82        without failing, which is why this is a construction error rather than a warning.
 83        """
 84        for field_name, template in (
 85            ("stop_condition", self.stop_condition),
 86            ("cursor_value", self.cursor_value),
 87        ):
 88            expression = template.string if isinstance(template, InterpolatedString) else template
 89            if isinstance(template, InterpolatedBoolean):
 90                expression = template.condition
 91            if not isinstance(expression, str) or not references_requested_page_size(expression):
 92                continue
 93            raise ValueError(
 94                f"The `{field_name}` {expression!r} reads the `{REQUESTED_PAGE_SIZE_VARIABLE}` interpolation "
 95                f"variable, but the CursorPagination strategy it belongs to declares no `page_size`, so there "
 96                f"is nothing to bind it to. The comparison would not fail either - it renders as the template "
 97                f"string itself, which is truthy - so the pagination would end after the first page and the "
 98                f"rest of the partition would be dropped silently. Declare `page_size` on the pagination "
 99                f"strategy, or compare against a value the manifest defines, such as "
100                f"`config['page_size']`."
101            )
102
103    @property
104    def initial_token(self) -> Optional[Any]:
105        """
106        CursorPaginationStrategy does not have an initial value because the next cursor is typically included
107        in the response of the first request. For Resumable Full Refresh streams that checkpoint the page
108        cursor, the next cursor should be read from the state or stream slice object.
109        """
110        return None
111
112    def next_page_token(
113        self,
114        response: requests.Response,
115        last_page_size: int,
116        last_record: Optional[Record],
117        last_page_token_value: Optional[Any] = None,
118        page_size_override: Optional[int] = None,
119        stream_slice: Optional[StreamSlice] = None,
120    ) -> Optional[Any]:
121        # The next page is a cursor read from the response, so `page_size_override` does not change how the token
122        # is computed. It is still exposed to the interpolation context as `page_size` because a `stop_condition`
123        # comparing `last_page_size` to a hardcoded page size would read a full reduced page as a short page and
124        # end the pagination early, silently dropping the rest of the partition. Writing the condition as
125        # `{{ last_page_size < page_size }}` keeps it correct while a reduction is in effect.
126        requested_page_size = (
127            page_size_override if page_size_override is not None else self._page_size
128        )
129        decoded_response = next(self.decoder.decode(response))
130        # The default way that link is presented in requests.Response is a string of various links (last, next, etc). This
131        # is not indexable or useful for parsing the cursor, so we replace it with the link dictionary from response.links
132        headers: Dict[str, Any] = dict(response.headers)
133        headers["link"] = response.links
134        if self._stop_condition:
135            should_stop = self._stop_condition.eval(
136                self.config,
137                response=decoded_response,
138                headers=headers,
139                last_record=last_record,
140                last_page_size=last_page_size,
141                page_size=requested_page_size,
142                stream_slice=stream_slice,
143            )
144            if should_stop:
145                return None
146        token = self._cursor_value.eval(
147            config=self.config,
148            response=decoded_response,
149            headers=headers,
150            last_record=last_record,
151            last_page_size=last_page_size,
152            page_size=requested_page_size,
153            stream_slice=stream_slice,
154        )
155        return token if token else None
156
157    def get_page_size(self) -> Optional[int]:
158        return self._page_size

33class CursorStopCondition(PaginationStopCondition):
34    def __init__(
35        self,
36        cursor: Cursor,
37    ):
38        self._cursor = cursor
39
40    def is_met(self, record: Record) -> bool:
41        return not self._cursor.should_be_synced(record)

 25@dataclass
 26class OffsetIncrement(PaginationStrategy):
 27    """
 28    Pagination strategy that returns the number of records reads so far and returns it as the next page token
 29    Examples:
 30        # page_size to be a constant integer value
 31        pagination_strategy:
 32          type: OffsetIncrement
 33          page_size: 2
 34
 35        # page_size to be a constant string value
 36        pagination_strategy:
 37          type: OffsetIncrement
 38          page_size: "2"
 39
 40        # page_size to be an interpolated string value
 41        pagination_strategy:
 42          type: OffsetIncrement
 43          page_size: "{{ parameters['items_per_page'] }}"
 44
 45    Attributes:
 46        page_size (InterpolatedString): the number of records to request
 47        extractor (Optional[RecordExtractor]): counts the records of a page by extracting them
 48            again. Leave it unset to use the count of the `RecordSelector` that read the page
 49    """
 50
 51    config: Config
 52    page_size: Optional[Union[str, int]]
 53    parameters: InitVar[Mapping[str, Any]]
 54    extractor: Optional[RecordExtractor] = None
 55    decoder: Decoder = field(
 56        default_factory=lambda: PaginationDecoderDecorator(decoder=JsonDecoder(parameters={}))
 57    )
 58    inject_on_first_request: bool = False
 59
 60    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 61        page_size = str(self.page_size) if isinstance(self.page_size, int) else self.page_size
 62        if page_size:
 63            self._page_size: Optional[InterpolatedString] = InterpolatedString(
 64                page_size, parameters=parameters
 65            )
 66        else:
 67            self._page_size = None
 68
 69    @property
 70    def initial_token(self) -> Optional[Any]:
 71        if self.inject_on_first_request:
 72            return 0
 73        return None
 74
 75    def next_page_token(
 76        self,
 77        response: requests.Response,
 78        last_page_size: int,
 79        last_record: Optional[Record],
 80        last_page_token_value: Optional[Any] = None,
 81        page_size_override: Optional[int] = None,
 82        stream_slice: Optional[StreamSlice] = None,
 83    ) -> Optional[Any]:
 84        # The records the API returned, not the ones emitted: a record filter must neither end the
 85        # pagination nor shorten the offset.
 86        if self.extractor:
 87            last_page_size = len(list(self.extractor.extract_records(response=response)))
 88        else:
 89            last_page_size = extracted_record_count(response, default=last_page_size)
 90
 91        # Stop paginating when there are fewer records than the page size or the current page has no records.
 92        # The comparison uses the page size that was actually requested: after a `REDUCE_PAGE_SIZE` reduction, a
 93        # full page is smaller than the configured page size and comparing against the latter would end the
 94        # pagination early and skip records.
 95        if page_size_override is not None:
 96            requested_page_size: Optional[int] = page_size_override
 97        elif isinstance(self.page_size, int):
 98            # A constant needs no response, which would otherwise be decoded a second time.
 99            requested_page_size = self.page_size
100        elif self._page_size:
101            requested_page_size = self._page_size.eval(
102                self.config, response=next(self.decoder.decode(response))
103            )
104        else:
105            requested_page_size = None
106        if (requested_page_size and last_page_size < requested_page_size) or last_page_size == 0:
107            return None
108        elif last_page_token_value is None:
109            # If the OffsetIncrement strategy does not inject on the first request, the incoming last_page_token_value
110            # will be None. For this case, we assume that None was the first page and progress to the next offset
111            return 0 + last_page_size
112        elif not isinstance(last_page_token_value, int):
113            raise ValueError(
114                f"Last page token value {last_page_token_value} for OffsetIncrement pagination strategy was not an integer"
115            )
116        else:
117            return last_page_token_value + last_page_size
118
119    def get_page_size(self) -> Optional[int]:
120        if self._page_size:
121            page_size = self._page_size.eval(self.config)
122            if not isinstance(page_size, int):
123                raise Exception(f"{page_size} is of type {type(page_size)}. Expected {int}")
124            return page_size
125        else:
126            return None

 22@dataclass
 23class PageIncrement(PaginationStrategy):
 24    """
 25    Pagination strategy that returns the number of pages reads so far and returns it as the next page token
 26
 27    Attributes:
 28        page_size (int): the number of records to request
 29        start_from_page (int): number of the initial page
 30        extractor (Optional[RecordExtractor]): counts the records of a page by extracting them
 31            again. Leave it unset to use the count of the `RecordSelector` that read the page
 32    """
 33
 34    config: Config
 35    page_size: Optional[Union[str, int]]
 36    parameters: InitVar[Mapping[str, Any]]
 37    start_from_page: int = 0
 38    inject_on_first_request: bool = False
 39    extractor: Optional[RecordExtractor] = None
 40
 41    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 42        if isinstance(self.page_size, int) or (self.page_size is None):
 43            self._page_size = self.page_size
 44        else:
 45            page_size = InterpolatedString(self.page_size, parameters=parameters).eval(self.config)
 46            if not isinstance(page_size, int):
 47                raise Exception(f"{page_size} is of type {type(page_size)}. Expected {int}")
 48            self._page_size = page_size
 49
 50    @property
 51    def initial_token(self) -> Optional[Any]:
 52        if self.inject_on_first_request:
 53            return self.start_from_page
 54        return None
 55
 56    def next_page_token(
 57        self,
 58        response: requests.Response,
 59        last_page_size: int,
 60        last_record: Optional[Record],
 61        last_page_token_value: Optional[Any],
 62        page_size_override: Optional[int] = None,
 63        stream_slice: Optional[StreamSlice] = None,
 64    ) -> Optional[Any]:
 65        if page_size_override is not None:
 66            # Reachable only when the factory is bypassed: a manifest naming PageIncrement, and a
 67            # CustomPaginationStrategy that subclasses it, are both rejected at config time. An unrecognized
 68            # ValueError here would be reported as a generic system error even though the message describes a
 69            # configuration mistake. `next_page_token` runs after the page's records have been emitted, so an
 70            # out-of-tree caller gets this mid-stream rather than at startup - still better than silently
 71            # ignoring the override and skipping records.
 72            raise AirbyteTracedException(
 73                internal_message="PageIncrement received a page_size_override",
 74                message="PageIncrement does not support reducing the page size while paginating: pages are "
 75                "addressed as page number * page size, so a smaller page size shifts every following page "
 76                "boundary and would skip records. Use OffsetIncrement or CursorPagination instead.",
 77                failure_type=FailureType.config_error,
 78            )
 79
 80        # The record count is dependent on the records returned from the response which may not always
 81        # align with the size of pages emitted. For example, a record filter can reduce the number of
 82        # records observed below the page size even though the API returned a full page.
 83        if self.extractor:
 84            last_page_size = len(list(self.extractor.extract_records(response=response)))
 85        else:
 86            last_page_size = extracted_record_count(response, default=last_page_size)
 87
 88        # Stop paginating when there are fewer records than the page size or the current page has no records
 89        if (self._page_size and last_page_size < self._page_size) or last_page_size == 0:
 90            return None
 91        elif last_page_token_value is None:
 92            # If the PageIncrement strategy does not inject on the first request, the incoming last_page_token_value
 93            # may be None. When this is the case, we assume we've already requested the first page specified by
 94            # start_from_page and must now get the next page
 95            return self.start_from_page + 1
 96        elif not isinstance(last_page_token_value, int):
 97            raise ValueError(
 98                f"Last page token value {last_page_token_value} for PageIncrement pagination strategy was not an integer"
 99            )
100        else:
101            return last_page_token_value + 1
102
103    def get_page_size(self) -> Optional[int]:
104        return self._page_size

44class StopConditionPaginationStrategyDecorator(PaginationStrategy):
45    def __init__(self, _delegate: PaginationStrategy, stop_condition: PaginationStopCondition):
46        self._delegate = _delegate
47        self._stop_condition = stop_condition
48
49    def next_page_token(
50        self,
51        response: requests.Response,
52        last_page_size: int,
53        last_record: Optional[Record],
54        last_page_token_value: Optional[Any] = None,
55        page_size_override: Optional[int] = None,
56        stream_slice: Optional[StreamSlice] = None,
57    ) -> Optional[Any]:
58        # We evaluate in reverse order because the assumption is that most of the APIs using data feed structure
59        # will return records in descending order. In terms of performance/memory, we return the records lazily
60        if last_record and self._stop_condition.is_met(last_record):
61            return None
62        return self._delegate.next_page_token(
63            response,
64            last_page_size,
65            last_record,
66            last_page_token_value,
67            **page_size_override_kwargs(page_size_override),
68            **stream_slice_kwargs(self._delegate.next_page_token, stream_slice),
69        )
70
71    def get_page_size(self) -> Optional[int]:
72        return self._delegate.get_page_size()
73
74    @property
75    def initial_token(self) -> Optional[Any]:
76        return self._delegate.initial_token