airbyte_cdk.sources.declarative.requesters.paginators.paginator

 1#
 2# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 3#
 4
 5from abc import ABC, abstractmethod
 6from dataclasses import dataclass
 7from typing import Any, Mapping, Optional
 8
 9import requests
10
11from airbyte_cdk.sources.declarative.requesters.request_options.request_options_provider import (
12    RequestOptionsProvider,
13)
14from airbyte_cdk.sources.types import Record, StreamSlice
15
16
17@dataclass
18class Paginator(ABC, RequestOptionsProvider):
19    """
20    Defines the token to use to fetch the next page of records from the API.
21
22    If needed, the Paginator will set request options to be set on the HTTP request to fetch the next page of records.
23    If the next_page_token is the path to the next page of records, then it should be accessed through the `path` method
24    """
25
26    @abstractmethod
27    def get_initial_token(self) -> Optional[Any]:
28        """
29        Get the page token that should be included in the request to get the first page of records
30        """
31
32    @abstractmethod
33    def next_page_token(
34        self,
35        response: requests.Response,
36        last_page_size: int,
37        last_record: Optional[Record],
38        last_page_token_value: Optional[Any],
39    ) -> Optional[Mapping[str, Any]]:
40        """
41        Returns the next_page_token to use to fetch the next page of records.
42
43        :param response: the response to process
44        :param last_page_size: the number of records read from the response
45        :param last_record: the last record extracted from the response
46        :param last_page_token_value: The current value of the page token made on the last request
47        :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.
48        """
49        pass
50
51    @abstractmethod
52    def path(
53        self,
54        next_page_token: Optional[Mapping[str, Any]],
55        stream_state: Optional[Mapping[str, Any]] = None,
56        stream_slice: Optional[StreamSlice] = None,
57    ) -> Optional[str]:
58        """
59        Returns the URL path to hit to fetch the next page of records
60
61        e.g: if you wanted to hit https://myapi.com/v1/some_entity then this will return "some_entity"
62
63        :return: path to hit to fetch the next request. Returning None means the path is not defined by the next_page_token
64        """
65        pass