airbyte_cdk.sources.declarative.requesters.paginators.strategies.stop_condition 

 1#
 2# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 3#
 4
 5from abc import ABC, abstractmethod
 6from typing import Any, Optional
 7
 8import requests
 9
10from airbyte_cdk.sources.declarative.requesters.paginators.paginator import (
11    page_size_override_kwargs,
12    stream_slice_kwargs,
13)
14from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import (
15    PaginationStrategy,
16)
17from airbyte_cdk.sources.streams.concurrent.cursor import Cursor
18from airbyte_cdk.sources.types import Record, StreamSlice
19
20
21class PaginationStopCondition(ABC):
22    @abstractmethod
23    def is_met(self, record: Record) -> bool:
24        """
25        Given a condition is met, the pagination will stop
26
27        :param record: a record used to evaluate the condition
28        """
29        raise NotImplementedError()
30
31
32class CursorStopCondition(PaginationStopCondition):
33    def __init__(
34        self,
35        cursor: Cursor,
36    ):
37        self._cursor = cursor
38
39    def is_met(self, record: Record) -> bool:
40        return not self._cursor.should_be_synced(record)
41
42
43class StopConditionPaginationStrategyDecorator(PaginationStrategy):
44    def __init__(self, _delegate: PaginationStrategy, stop_condition: PaginationStopCondition):
45        self._delegate = _delegate
46        self._stop_condition = stop_condition
47
48    def next_page_token(
49        self,
50        response: requests.Response,
51        last_page_size: int,
52        last_record: Optional[Record],
53        last_page_token_value: Optional[Any] = None,
54        page_size_override: Optional[int] = None,
55        stream_slice: Optional[StreamSlice] = None,
56    ) -> Optional[Any]:
57        # We evaluate in reverse order because the assumption is that most of the APIs using data feed structure
58        # will return records in descending order. In terms of performance/memory, we return the records lazily
59        if last_record and self._stop_condition.is_met(last_record):
60            return None
61        return self._delegate.next_page_token(
62            response,
63            last_page_size,
64            last_record,
65            last_page_token_value,
66            **page_size_override_kwargs(page_size_override),
67            **stream_slice_kwargs(self._delegate.next_page_token, stream_slice),
68        )
69
70    def get_page_size(self) -> Optional[int]:
71        return self._delegate.get_page_size()
72
73    @property
74    def initial_token(self) -> Optional[Any]:
75        return self._delegate.initial_token
class PaginationStopCondition(abc.ABC): 
22class PaginationStopCondition(ABC):
23    @abstractmethod
24    def is_met(self, record: Record) -> bool:
25        """
26        Given a condition is met, the pagination will stop
27
28        :param record: a record used to evaluate the condition
29        """
30        raise NotImplementedError()

Helper class that provides a standard way to create an ABC using inheritance.

@abstractmethod
def is_met(self, record: airbyte_cdk.Record) -> bool: 
23    @abstractmethod
24    def is_met(self, record: Record) -> bool:
25        """
26        Given a condition is met, the pagination will stop
27
28        :param record: a record used to evaluate the condition
29        """
30        raise NotImplementedError()

Given a condition is met, the pagination will stop

Parameters
  • record: a record used to evaluate the condition
class CursorStopCondition(PaginationStopCondition): 
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)

Helper class that provides a standard way to create an ABC using inheritance.

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

Given a condition is met, the pagination will stop

Parameters
  • record: a record used to evaluate the condition
class StopConditionPaginationStrategyDecorator(airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy.PaginationStrategy): 
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

Defines how to get the next page token

One instance paginates every partition of the stream, several at once, so next_page_token must derive the next token from its arguments, such as last_page_token_value, rather than from state kept on the instance.

StopConditionPaginationStrategyDecorator( _delegate: airbyte_cdk.PaginationStrategy, stop_condition: PaginationStopCondition) 
45    def __init__(self, _delegate: PaginationStrategy, stop_condition: PaginationStopCondition):
46        self._delegate = _delegate
47        self._stop_condition = stop_condition
def next_page_token( self, response: requests.models.Response, last_page_size: int, last_record: Optional[airbyte_cdk.Record], last_page_token_value: Optional[Any] = None, page_size_override: Optional[int] = None, stream_slice: Optional[airbyte_cdk.StreamSlice] = None) -> Optional[Any]: 
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        )
Parameters
  • response: response to process
  • last_page_size: the number of records read from the response
  • last_record: the last record extracted from the response
  • last_page_token_value: The current value of the page token made on the last request
  • page_size_override: the page size that was actually requested, when it differs from the configured one because of a REDUCE_PAGE_SIZE response action. Strategies that use their configured page size as a stop condition must honor this value, else they end the pagination early and skip records.
  • stream_slice: the slice the page was read for. Only passed to strategies that declare it, so existing strategies defined outside of the CDK keep working unchanged.
Returns

next page token. Returns None if there are no more pages to fetch

def get_page_size(self) -> Optional[int]: 
71    def get_page_size(self) -> Optional[int]:
72        return self._delegate.get_page_size()
Returns

page size: The number of records to fetch in a page. Returns None if unspecified

initial_token: Optional[Any] 
74    @property
75    def initial_token(self) -> Optional[Any]:
76        return self._delegate.initial_token

Return the initial value of the token