airbyte_cdk.sources.declarative.requesters.paginators

 37@dataclass
 38class DefaultPaginator(Paginator):
 39    """
 40    Default paginator to request pages of results with a fixed size until the pagination strategy no longer returns a next_page_token
 41
 42    Examples:
 43        1.
 44        * fetches up to 10 records at a time by setting the "limit" request param to 10
 45        * updates the request path with  "{{ response._metadata.next }}"
 46        ```
 47          paginator:
 48            type: "DefaultPaginator"
 49            page_size_option:
 50              type: RequestOption
 51              inject_into: request_parameter
 52              field_name: limit
 53            page_token_option:
 54              type: RequestPath
 55              path: "location"
 56            pagination_strategy:
 57              type: "CursorPagination"
 58              cursor_value: "{{ response._metadata.next }}"
 59              page_size: 10
 60        ```
 61
 62        2.
 63        * fetches up to 5 records at a time by setting the "page_size" header to 5
 64        * increments a record counter and set the request parameter "offset" to the value of the counter
 65        ```
 66          paginator:
 67            type: "DefaultPaginator"
 68            page_size_option:
 69              type: RequestOption
 70              inject_into: header
 71              field_name: page_size
 72            pagination_strategy:
 73              type: "OffsetIncrement"
 74              page_size: 5
 75            page_token_option:
 76              option_type: "request_parameter"
 77              field_name: "offset"
 78        ```
 79
 80        3.
 81        * fetches up to 5 records at a time by setting the "page_size" request param to 5
 82        * increments a page counter and set the request parameter "page" to the value of the counter
 83        ```
 84          paginator:
 85            type: "DefaultPaginator"
 86            page_size_option:
 87              type: RequestOption
 88              inject_into: request_parameter
 89              field_name: page_size
 90            pagination_strategy:
 91              type: "PageIncrement"
 92              page_size: 5
 93            page_token_option:
 94              type: RequestOption
 95              option_type: "request_parameter"
 96              field_name: "page"
 97        ```
 98    Attributes:
 99        page_size_option (Optional[RequestOption]): the request option to set the page size. Cannot be injected in the path.
100        page_token_option (Optional[RequestPath, RequestOption]): the request option to set the page token
101        pagination_strategy (PaginationStrategy): Strategy defining how to get the next page token
102        config (Config): connection config
103        url_base (Union[InterpolatedString, str]): endpoint's base url
104        decoder (Decoder): decoder to decode the response
105    """
106
107    pagination_strategy: PaginationStrategy
108    config: Config
109    url_base: Union[InterpolatedString, str]
110    parameters: InitVar[Mapping[str, Any]]
111    decoder: Decoder = field(
112        default_factory=lambda: PaginationDecoderDecorator(decoder=JsonDecoder(parameters={}))
113    )
114    page_size_option: Optional[RequestOption] = None
115    page_token_option: Optional[Union[RequestPath, RequestOption]] = None
116
117    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
118        if self.page_size_option and not self.pagination_strategy.get_page_size():
119            raise ValueError(
120                "page_size_option cannot be set if the pagination strategy does not have a page_size"
121            )
122        if isinstance(self.url_base, str):
123            self.url_base = InterpolatedString(string=self.url_base, parameters=parameters)
124
125        if self.page_token_option and not isinstance(self.page_token_option, RequestPath):
126            _validate_component_request_option_paths(
127                self.config,
128                self.page_size_option,
129                self.page_token_option,
130            )
131
132    def get_initial_token(self) -> Optional[Any]:
133        """
134        Return the page token that should be used for the first request of a stream
135
136        WARNING: get_initial_token() should not be used by streams that use RFR that perform checkpointing
137        of state using page numbers. Because paginators are stateless
138        """
139        return self.pagination_strategy.initial_token
140
141    def get_page_size(self) -> Optional[int]:
142        return self.pagination_strategy.get_page_size()
143
144    def next_page_token(
145        self,
146        response: requests.Response,
147        last_page_size: int,
148        last_record: Optional[Record],
149        last_page_token_value: Optional[Any] = None,
150        page_size_override: Optional[int] = None,
151        stream_slice: Optional[StreamSlice] = None,
152    ) -> Optional[Mapping[str, Any]]:
153        next_page_token = self.pagination_strategy.next_page_token(
154            response=response,
155            last_page_size=last_page_size,
156            last_record=last_record,
157            last_page_token_value=last_page_token_value,
158            **page_size_override_kwargs(page_size_override),
159            **stream_slice_kwargs(self.pagination_strategy.next_page_token, stream_slice),
160        )
161        if next_page_token:
162            return {"next_page_token": next_page_token}
163        else:
164            return None
165
166    def path(
167        self,
168        next_page_token: Optional[Mapping[str, Any]],
169        stream_state: Optional[Mapping[str, Any]] = None,
170        stream_slice: Optional[StreamSlice] = None,
171    ) -> Optional[str]:
172        token = next_page_token.get("next_page_token") if next_page_token else None
173        if token and self.page_token_option and isinstance(self.page_token_option, RequestPath):
174            return str(token)
175        else:
176            return None
177
178    def get_request_params(
179        self,
180        *,
181        stream_state: Optional[StreamState] = None,
182        stream_slice: Optional[StreamSlice] = None,
183        next_page_token: Optional[Mapping[str, Any]] = None,
184        page_size_override: Optional[int] = None,
185    ) -> MutableMapping[str, Any]:
186        return self._get_request_options(
187            RequestOptionType.request_parameter, next_page_token, page_size_override
188        )
189
190    def get_request_headers(
191        self,
192        *,
193        stream_state: Optional[StreamState] = None,
194        stream_slice: Optional[StreamSlice] = None,
195        next_page_token: Optional[Mapping[str, Any]] = None,
196        page_size_override: Optional[int] = None,
197    ) -> Mapping[str, str]:
198        return self._get_request_options(
199            RequestOptionType.header, next_page_token, page_size_override
200        )
201
202    def get_request_body_data(
203        self,
204        *,
205        stream_state: Optional[StreamState] = None,
206        stream_slice: Optional[StreamSlice] = None,
207        next_page_token: Optional[Mapping[str, Any]] = None,
208        page_size_override: Optional[int] = None,
209    ) -> Mapping[str, Any]:
210        return self._get_request_options(
211            RequestOptionType.body_data, next_page_token, page_size_override
212        )
213
214    def get_request_body_json(
215        self,
216        *,
217        stream_state: Optional[StreamState] = None,
218        stream_slice: Optional[StreamSlice] = None,
219        next_page_token: Optional[Mapping[str, Any]] = None,
220        page_size_override: Optional[int] = None,
221    ) -> Mapping[str, Any]:
222        return self._get_request_options(
223            RequestOptionType.body_json, next_page_token, page_size_override
224        )
225
226    def _get_request_options(
227        self,
228        option_type: RequestOptionType,
229        next_page_token: Optional[Mapping[str, Any]],
230        page_size_override: Optional[int] = None,
231    ) -> MutableMapping[str, Any]:
232        options: MutableMapping[str, Any] = {}
233
234        token = next_page_token.get("next_page_token") if next_page_token else None
235        if (
236            self.page_token_option
237            and token is not None
238            and isinstance(self.page_token_option, RequestOption)
239            and self.page_token_option.inject_into == option_type
240        ):
241            self.page_token_option.inject_into_request(options, token, self.config)
242
243        if self.page_size_option and self.page_size_option.inject_into == option_type:
244            page_size = (
245                page_size_override
246                if page_size_override is not None
247                else self.pagination_strategy.get_page_size()
248            )
249            if page_size:
250                self.page_size_option.inject_into_request(options, page_size, self.config)
251
252        return options

15@dataclass
16class NoPagination(Paginator):
17    """
18    Pagination implementation that never returns a next page.
19    """
20
21    parameters: InitVar[Mapping[str, Any]]
22
23    def path(
24        self,
25        next_page_token: Optional[Mapping[str, Any]],
26        stream_state: Optional[Mapping[str, Any]] = None,
27        stream_slice: Optional[StreamSlice] = None,
28    ) -> Optional[str]:
29        return None
30
31    def get_request_params(
32        self,
33        *,
34        stream_state: Optional[StreamState] = None,
35        stream_slice: Optional[StreamSlice] = None,
36        next_page_token: Optional[Mapping[str, Any]] = None,
37    ) -> MutableMapping[str, Any]:
38        return {}
39
40    def get_request_headers(
41        self,
42        *,
43        stream_state: Optional[StreamState] = None,
44        stream_slice: Optional[StreamSlice] = None,
45        next_page_token: Optional[Mapping[str, Any]] = None,
46    ) -> Mapping[str, str]:
47        return {}
48
49    def get_request_body_data(
50        self,
51        *,
52        stream_state: Optional[StreamState] = None,
53        stream_slice: Optional[StreamSlice] = None,
54        next_page_token: Optional[Mapping[str, Any]] = None,
55    ) -> Union[Mapping[str, Any], str]:
56        return {}
57
58    def get_request_body_json(
59        self,
60        *,
61        stream_state: Optional[StreamState] = None,
62        stream_slice: Optional[StreamSlice] = None,
63        next_page_token: Optional[Mapping[str, Any]] = None,
64    ) -> Mapping[str, Any]:
65        return {}
66
67    def get_initial_token(self) -> Optional[Any]:
68        return None
69
70    def next_page_token(
71        self,
72        response: requests.Response,
73        last_page_size: int,
74        last_record: Optional[Record],
75        last_page_token_value: Optional[Any],
76        page_size_override: Optional[int] = None,
77        stream_slice: Optional[StreamSlice] = None,
78    ) -> Optional[Mapping[str, Any]]:
79        return {}

15@dataclass
16class PaginationStrategy:
17    """
18    Defines how to get the next page token
19
20    One instance paginates every partition of the stream, several at once, so `next_page_token` must derive the
21    next token from its arguments, such as `last_page_token_value`, rather than from state kept on the instance.
22    """
23
24    @property
25    @abstractmethod
26    def initial_token(self) -> Optional[Any]:
27        """
28        Return the initial value of the token
29        """
30
31    @abstractmethod
32    def next_page_token(
33        self,
34        response: requests.Response,
35        last_page_size: int,
36        last_record: Optional[Record],
37        last_page_token_value: Optional[Any],
38        page_size_override: Optional[int] = None,
39        stream_slice: Optional[StreamSlice] = None,
40    ) -> Optional[Any]:
41        """
42        :param response: response to process
43        :param last_page_size: the number of records read from the response
44        :param last_record: the last record extracted from the response
45        :param last_page_token_value: The current value of the page token made on the last request
46        :param page_size_override: the page size that was actually requested, when it differs from the configured
47            one because of a `REDUCE_PAGE_SIZE` response action. Strategies that use their configured page size as
48            a stop condition must honor this value, else they end the pagination early and skip records.
49        :param stream_slice: the slice the page was read for. Only passed to strategies that declare it, so existing
50            strategies defined outside of the CDK keep working unchanged.
51        :return: next page token. Returns None if there are no more pages to fetch
52        """
53        pass
54
55    @abstractmethod
56    def get_page_size(self) -> Optional[int]:
57        """
58        :return: page size: The number of records to fetch in a page. Returns None if unspecified
59        """

 63@dataclass
 64class Paginator(ABC, RequestOptionsProvider):
 65    """
 66    Defines the token to use to fetch the next page of records from the API.
 67
 68    If needed, the Paginator will set request options to be set on the HTTP request to fetch the next page of records.
 69    If the next_page_token is the path to the next page of records, then it should be accessed through the `path` method
 70    """
 71
 72    @abstractmethod
 73    def get_initial_token(self) -> Optional[Any]:
 74        """
 75        Get the page token that should be included in the request to get the first page of records
 76        """
 77
 78    @abstractmethod
 79    def next_page_token(
 80        self,
 81        response: requests.Response,
 82        last_page_size: int,
 83        last_record: Optional[Record],
 84        last_page_token_value: Optional[Any],
 85        page_size_override: Optional[int] = None,
 86        stream_slice: Optional[StreamSlice] = None,
 87    ) -> Optional[Mapping[str, Any]]:
 88        """
 89        Returns the next_page_token to use to fetch the next page of records.
 90
 91        :param response: the response to process
 92        :param last_page_size: the number of records read from the response
 93        :param last_record: the last record extracted from the response
 94        :param last_page_token_value: The current value of the page token made on the last request
 95        :param page_size_override: the page size that was actually requested, when it differs from the configured
 96            one because of a `REDUCE_PAGE_SIZE` response action
 97        :param stream_slice: the slice the page was read for, so that a stop condition can compare the page against
 98            the slice's own window or partition
 99        :return: A mapping {"next_page_token": <token>} for the next page from the input response object. Returning None means there are no more pages to read in this response.
100        """
101        pass
102
103    def get_page_size(self) -> Optional[int]:
104        """
105        Evaluated against the config alone. A pagination strategy whose `page_size` template references the
106        response evaluates it with the response in context inside `next_page_token`, so the two can disagree -
107        pre-existing, and only observable for a `page_size` that is not a constant.
108
109        :return: the number of records this paginator asks for per page, or None if it does not define one
110        """
111        return None
112
113    @abstractmethod
114    def path(
115        self,
116        next_page_token: Optional[Mapping[str, Any]],
117        stream_state: Optional[Mapping[str, Any]] = None,
118        stream_slice: Optional[StreamSlice] = None,
119    ) -> Optional[str]:
120        """
121        Returns the URL path to hit to fetch the next page of records
122
123        e.g: if you wanted to hit https://myapi.com/v1/some_entity then this will return "some_entity"
124
125        :return: path to hit to fetch the next request. Returning None means the path is not defined by the next_page_token
126        """
127        pass

255class PaginatorTestReadDecorator(Paginator):
256    """
257    In some cases, we want to limit the number of requests that are made to the backend source. This class allows for limiting the number of
258    pages that are queried throughout a read command.
259
260    WARNING: This decorator is not currently thread-safe like the rest of the low-code framework because it has
261    an internal state to track the current number of pages counted so that it can exit early during a test read
262
263    The count bounds *successful* pages: it is incremented from `next_page_token`, which a page that resolved
264    to `ResponseAction.REDUCE_PAGE_SIZE` never reaches because `SimpleRetriever._read_pages` re-issues that
265    page before asking for the next token. A stream with `page_size_reduction` can therefore issue up to
266    `max_attempts` extra requests per slice on top of this limit.
267
268    The Connector Builder counts pages a second time, from the request/response logs rather than from this
269    class (`connector_builder/test_reader/reader.py::_has_reached_limit`). The two counts agree because
270    `HttpClient` logs a response resolving to `REDUCE_PAGE_SIZE` as an auxiliary request, which the Builder
271    does not turn into a page.
272
273    `get_initial_token()` also resets this count, and `SimpleRetriever._read_pages` calls it again from scratch
274    for every child window a `SPLIT_REQUEST_WINDOW` response produces. A split slice can therefore fetch up to
275    `maximum_number_of_pages` per child rather than sharing one budget across the whole original partition.
276    """
277
278    _PAGE_COUNT_BEFORE_FIRST_NEXT_CALL = 1
279
280    def __init__(self, decorated: Paginator, maximum_number_of_pages: int = 5) -> None:
281        if maximum_number_of_pages and maximum_number_of_pages < 1:
282            raise ValueError(
283                f"The maximum number of pages on a test read needs to be strictly positive. Got {maximum_number_of_pages}"
284            )
285        self._maximum_number_of_pages = maximum_number_of_pages
286        self._decorated = decorated
287        self._page_count = self._PAGE_COUNT_BEFORE_FIRST_NEXT_CALL
288
289    def get_initial_token(self) -> Optional[Any]:
290        self._page_count = self._PAGE_COUNT_BEFORE_FIRST_NEXT_CALL
291        return self._decorated.get_initial_token()
292
293    def get_page_size(self) -> Optional[int]:
294        return self._decorated.get_page_size()
295
296    def next_page_token(
297        self,
298        response: requests.Response,
299        last_page_size: int,
300        last_record: Optional[Record],
301        last_page_token_value: Optional[Any] = None,
302        page_size_override: Optional[int] = None,
303        stream_slice: Optional[StreamSlice] = None,
304    ) -> Optional[Mapping[str, Any]]:
305        if self._page_count >= self._maximum_number_of_pages:
306            return None
307
308        self._page_count += 1
309        return self._decorated.next_page_token(
310            response,
311            last_page_size,
312            last_record,
313            last_page_token_value,
314            **page_size_override_kwargs(page_size_override),
315            **stream_slice_kwargs(self._decorated.next_page_token, stream_slice),
316        )
317
318    def path(
319        self,
320        next_page_token: Optional[Mapping[str, Any]],
321        stream_state: Optional[Mapping[str, Any]] = None,
322        stream_slice: Optional[StreamSlice] = None,
323    ) -> Optional[str]:
324        return self._decorated.path(
325            next_page_token=next_page_token,
326            stream_state=stream_state,
327            stream_slice=stream_slice,
328        )
329
330    def get_request_params(
331        self,
332        *,
333        stream_state: Optional[StreamState] = None,
334        stream_slice: Optional[StreamSlice] = None,
335        next_page_token: Optional[Mapping[str, Any]] = None,
336        page_size_override: Optional[int] = None,
337    ) -> Mapping[str, Any]:
338        return self._decorated.get_request_params(
339            stream_state=stream_state,
340            stream_slice=stream_slice,
341            next_page_token=next_page_token,
342            **page_size_override_kwargs(page_size_override),
343        )
344
345    def get_request_headers(
346        self,
347        *,
348        stream_state: Optional[StreamState] = None,
349        stream_slice: Optional[StreamSlice] = None,
350        next_page_token: Optional[Mapping[str, Any]] = None,
351        page_size_override: Optional[int] = None,
352    ) -> Mapping[str, str]:
353        return self._decorated.get_request_headers(
354            stream_state=stream_state,
355            stream_slice=stream_slice,
356            next_page_token=next_page_token,
357            **page_size_override_kwargs(page_size_override),
358        )
359
360    def get_request_body_data(
361        self,
362        *,
363        stream_state: Optional[StreamState] = None,
364        stream_slice: Optional[StreamSlice] = None,
365        next_page_token: Optional[Mapping[str, Any]] = None,
366        page_size_override: Optional[int] = None,
367    ) -> Union[Mapping[str, Any], str]:
368        return self._decorated.get_request_body_data(
369            stream_state=stream_state,
370            stream_slice=stream_slice,
371            next_page_token=next_page_token,
372            **page_size_override_kwargs(page_size_override),
373        )
374
375    def get_request_body_json(
376        self,
377        *,
378        stream_state: Optional[StreamState] = None,
379        stream_slice: Optional[StreamSlice] = None,
380        next_page_token: Optional[Mapping[str, Any]] = None,
381        page_size_override: Optional[int] = None,
382    ) -> Mapping[str, Any]:
383        return self._decorated.get_request_body_json(
384            stream_state=stream_state,
385            stream_slice=stream_slice,
386            next_page_token=next_page_token,
387            **page_size_override_kwargs(page_size_override),
388        )