From a274d90ea9e98f48fbf035099b3064774ecdc4d9 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Wed, 23 Sep 2026 18:09:28 +0300 Subject: [PATCH] feat(low-code): expose `stream_slice` to the paginator interpolation context Pass the slice being read through `SimpleRetriever` and `DefaultPaginator` into `PaginationStrategy.next_page_token`, and add it (with the `stream_interval` / `stream_partition` aliases) to the context of `CursorPagination.stop_condition` and `cursor_value`. Paginators and strategies defined outside the CDK are only handed the slice when their `next_page_token` declares `stream_slice` or `**kwargs`, so existing custom strategies keep working unchanged. Closes #1166 Co-Authored-By: Claude Opus 5.5 --- .../declarative_component_schema.yaml | 6 ++ .../paginators/default_paginator.py | 5 ++ .../requesters/paginators/no_pagination.py | 1 + .../requesters/paginators/paginator.py | 40 ++++++++++++- .../strategies/cursor_pagination_strategy.py | 5 +- .../paginators/strategies/offset_increment.py | 3 +- .../paginators/strategies/page_increment.py | 3 +- .../strategies/pagination_strategy.py | 5 +- .../paginators/strategies/stop_condition.py | 5 +- .../retrievers/simple_retriever.py | 18 +++++- .../test_cursor_pagination_strategy.py | 60 ++++++++++++++++++- .../paginators/test_default_paginator.py | 49 +++++++++++++++ .../paginators/test_stop_condition.py | 18 +++++- .../retrievers/test_simple_retriever.py | 56 +++++++++++++++++ 14 files changed, 264 insertions(+), 10 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index a02be6fa0e..56d373afa2 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -754,6 +754,9 @@ definitions: - last_record - page_size - response + - stream_interval + - stream_partition + - stream_slice examples: - "{{ headers.link.next.cursor }}" - "{{ last_record['key'] }}" @@ -789,6 +792,9 @@ definitions: - last_record - page_size - response + - stream_interval + - stream_partition + - stream_slice examples: - "{{ response.data.has_more is false }}" - "{{ 'next' not in headers['link'] }}" diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py b/airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py index d00e75ad29..839a4a678a 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py @@ -16,6 +16,7 @@ from airbyte_cdk.sources.declarative.requesters.paginators.paginator import ( Paginator, page_size_override_kwargs, + stream_slice_kwargs, ) from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import ( PaginationStrategy, @@ -146,6 +147,7 @@ def next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any] = None, page_size_override: Optional[int] = None, + stream_slice: Optional[StreamSlice] = None, ) -> Optional[Mapping[str, Any]]: next_page_token = self.pagination_strategy.next_page_token( response=response, @@ -153,6 +155,7 @@ def next_page_token( last_record=last_record, last_page_token_value=last_page_token_value, **page_size_override_kwargs(page_size_override), + **stream_slice_kwargs(self.pagination_strategy.next_page_token, stream_slice), ) if next_page_token: return {"next_page_token": next_page_token} @@ -292,6 +295,7 @@ def next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any] = None, page_size_override: Optional[int] = None, + stream_slice: Optional[StreamSlice] = None, ) -> Optional[Mapping[str, Any]]: if self._page_count >= self._maximum_number_of_pages: return None @@ -303,6 +307,7 @@ def next_page_token( last_record, last_page_token_value, **page_size_override_kwargs(page_size_override), + **stream_slice_kwargs(self._decorated.next_page_token, stream_slice), ) def path( diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/no_pagination.py b/airbyte_cdk/sources/declarative/requesters/paginators/no_pagination.py index f207b1cce9..6a39851cde 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/no_pagination.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/no_pagination.py @@ -73,5 +73,6 @@ def next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any], page_size_override: Optional[int] = None, + stream_slice: Optional[StreamSlice] = None, ) -> Optional[Mapping[str, Any]]: return {} diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/paginator.py b/airbyte_cdk/sources/declarative/requesters/paginators/paginator.py index 17f974416e..dd5f095045 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/paginator.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/paginator.py @@ -2,9 +2,11 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # +import inspect from abc import ABC, abstractmethod from dataclasses import dataclass -from typing import Any, Dict, Mapping, Optional +from functools import lru_cache +from typing import Any, Callable, Dict, Mapping, Optional import requests @@ -24,6 +26,39 @@ def page_size_override_kwargs(page_size_override: Optional[int]) -> Dict[str, An return {"page_size_override": page_size_override} if page_size_override is not None else {} +def stream_slice_kwargs( + next_page_token: Callable[..., Any], stream_slice: Optional[StreamSlice] +) -> Dict[str, Any]: + """ + Build the `stream_slice` keyword argument only for a `next_page_token` that accepts it. + + Unlike `page_size_override`, the slice is almost always present, so passing it only when it is set would still + break a paginator or pagination strategy defined outside of the CDK whose signature predates the argument. The + signature is checked instead, and a callee that does not declare `stream_slice` (or `**kwargs`) is called as + before. + """ + if stream_slice is None: + return {} + function = getattr(next_page_token, "__func__", next_page_token) + try: + accepts_stream_slice = _accepts_stream_slice(function) + except TypeError: + # Unhashable callables cannot be cached; they are rare enough to inspect on every call. + accepts_stream_slice = _accepts_stream_slice.__wrapped__(function) + return {"stream_slice": stream_slice} if accepts_stream_slice else {} + + +@lru_cache(maxsize=None) +def _accepts_stream_slice(function: Callable[..., Any]) -> bool: + try: + parameters = inspect.signature(function).parameters + except (TypeError, ValueError): + return False + return "stream_slice" in parameters or any( + parameter.kind == inspect.Parameter.VAR_KEYWORD for parameter in parameters.values() + ) + + @dataclass class Paginator(ABC, RequestOptionsProvider): """ @@ -47,6 +82,7 @@ def next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any], page_size_override: Optional[int] = None, + stream_slice: Optional[StreamSlice] = None, ) -> Optional[Mapping[str, Any]]: """ Returns the next_page_token to use to fetch the next page of records. @@ -57,6 +93,8 @@ def next_page_token( :param last_page_token_value: The current value of the page token made on the last request :param 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 + :param stream_slice: the slice the page was read for, so that a stop condition can compare the page against + the slice's own window or partition :return: A mapping {"next_page_token": } for the next page from the input response object. Returning None means there are no more pages to read in this response. """ pass diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py index a4c43aed45..5d33981317 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py @@ -21,7 +21,7 @@ from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import ( PaginationStrategy, ) -from airbyte_cdk.sources.types import Config, Record +from airbyte_cdk.sources.types import Config, Record, StreamSlice @dataclass @@ -115,6 +115,7 @@ def next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any] = None, page_size_override: Optional[int] = None, + stream_slice: Optional[StreamSlice] = None, ) -> Optional[Any]: # The next page is a cursor read from the response, so `page_size_override` does not change how the token # is computed. It is still exposed to the interpolation context as `page_size` because a `stop_condition` @@ -137,6 +138,7 @@ def next_page_token( last_record=last_record, last_page_size=last_page_size, page_size=requested_page_size, + stream_slice=stream_slice, ) if should_stop: return None @@ -147,6 +149,7 @@ def next_page_token( last_record=last_record, last_page_size=last_page_size, page_size=requested_page_size, + stream_slice=stream_slice, ) return token if token else None diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/offset_increment.py b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/offset_increment.py index 46939402c4..a510731a2e 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/offset_increment.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/offset_increment.py @@ -17,7 +17,7 @@ from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import ( PaginationStrategy, ) -from airbyte_cdk.sources.types import Config, Record +from airbyte_cdk.sources.types import Config, Record, StreamSlice @dataclass @@ -75,6 +75,7 @@ def next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any] = None, page_size_override: Optional[int] = None, + stream_slice: Optional[StreamSlice] = None, ) -> Optional[Any]: decoded_response = next(self.decoder.decode(response)) diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/page_increment.py b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/page_increment.py index 026e0452d1..61b5e94f70 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/page_increment.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/page_increment.py @@ -13,7 +13,7 @@ from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import ( PaginationStrategy, ) -from airbyte_cdk.sources.types import Config, Record +from airbyte_cdk.sources.types import Config, Record, StreamSlice from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -56,6 +56,7 @@ def next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any], page_size_override: Optional[int] = None, + stream_slice: Optional[StreamSlice] = None, ) -> Optional[Any]: if page_size_override is not None: # Reachable only when the factory is bypassed: a manifest naming PageIncrement, and a diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/pagination_strategy.py b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/pagination_strategy.py index 9be3af80ed..a7a89f5400 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/pagination_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/pagination_strategy.py @@ -8,7 +8,7 @@ import requests -from airbyte_cdk.sources.types import Record +from airbyte_cdk.sources.types import Record, StreamSlice @dataclass @@ -32,6 +32,7 @@ def next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any], page_size_override: Optional[int] = None, + stream_slice: Optional[StreamSlice] = None, ) -> Optional[Any]: """ :param response: response to process @@ -41,6 +42,8 @@ def next_page_token( :param 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. + :param 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. :return: next page token. Returns None if there are no more pages to fetch """ pass diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py index 6df7a3a344..9fed018842 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py @@ -9,12 +9,13 @@ from airbyte_cdk.sources.declarative.requesters.paginators.paginator import ( page_size_override_kwargs, + stream_slice_kwargs, ) from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import ( PaginationStrategy, ) from airbyte_cdk.sources.streams.concurrent.cursor import Cursor -from airbyte_cdk.sources.types import Record +from airbyte_cdk.sources.types import Record, StreamSlice class PaginationStopCondition(ABC): @@ -51,6 +52,7 @@ def next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any] = None, page_size_override: Optional[int] = None, + stream_slice: Optional[StreamSlice] = None, ) -> Optional[Any]: # We evaluate in reverse order because the assumption is that most of the APIs using data feed structure # will return records in descending order. In terms of performance/memory, we return the records lazily @@ -62,6 +64,7 @@ def next_page_token( last_record, last_page_token_value, **page_size_override_kwargs(page_size_override), + **stream_slice_kwargs(self._delegate.next_page_token, stream_slice), ) def get_page_size(self) -> Optional[int]: diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 85871d0eb7..21f56347a9 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -36,6 +36,7 @@ from airbyte_cdk.sources.declarative.requesters.paginators.paginator import ( Paginator, page_size_override_kwargs, + stream_slice_kwargs, ) from airbyte_cdk.sources.declarative.requesters.query_properties import QueryProperties from airbyte_cdk.sources.declarative.requesters.request_options import ( @@ -329,6 +330,7 @@ def _next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any], page_size_override: Optional[int] = None, + stream_slice: Optional[StreamSlice] = None, ) -> Optional[Mapping[str, Any]]: """ Specifies a pagination strategy. @@ -343,6 +345,7 @@ def _next_page_token( last_record=last_record, last_page_token_value=last_page_token_value, **page_size_override_kwargs(page_size_override), + **stream_slice_kwargs(self._paginator.next_page_token, stream_slice), ) def _fetch_next_page( @@ -518,6 +521,7 @@ def _read_pages( last_record=last_record, last_page_token_value=last_page_token_value, **page_size_override_kwargs(page_size_override), + **stream_slice_kwargs(self._next_page_token, stream_slice), ) if not next_page_token: break @@ -629,7 +633,13 @@ def _read_pages( last_record = record yield record - next_page_token = self._next_page_token(response, last_page_size, last_record, None) + next_page_token = self._next_page_token( + response, + last_page_size, + last_record, + None, + **stream_slice_kwargs(self._next_page_token, stream_slice), + ) if next_page_token: yield from self._paginate( next_page_token, @@ -669,7 +679,11 @@ def _paginate( next_page_token.get("next_page_token") if next_page_token else None ) next_page_token = self._next_page_token( - response, last_page_size, last_record, last_page_token_value + response, + last_page_size, + last_record, + last_page_token_value, + **stream_slice_kwargs(self._next_page_token, stream_slice), ) if not next_page_token: diff --git a/unit_tests/sources/declarative/requesters/paginators/test_cursor_pagination_strategy.py b/unit_tests/sources/declarative/requesters/paginators/test_cursor_pagination_strategy.py index 4b25086fa1..b5b6b73d99 100644 --- a/unit_tests/sources/declarative/requesters/paginators/test_cursor_pagination_strategy.py +++ b/unit_tests/sources/declarative/requesters/paginators/test_cursor_pagination_strategy.py @@ -12,7 +12,7 @@ from airbyte_cdk.sources.declarative.requesters.paginators.strategies.cursor_pagination_strategy import ( CursorPaginationStrategy, ) -from airbyte_cdk.sources.types import Record +from airbyte_cdk.sources.types import Record, StreamSlice @pytest.mark.parametrize( @@ -305,3 +305,61 @@ def test_given_a_page_size_when_an_expression_reads_page_size_then_accept(page_s assert strategy.next_page_token(response, 100, None, None) == "a token" assert strategy.next_page_token(response, 50, None, None, page_size_override=50) == "a token" assert strategy.next_page_token(response, 49, None, None, page_size_override=50) is None + + +@pytest.mark.parametrize( + "oldest_created_at, expected_token", + [ + pytest.param( + "2024-06-20T00:00:00Z", "page-2", id="test_oldest_run_inside_the_window_then_continue" + ), + pytest.param( + "2024-05-30T00:00:00Z", + "page-2", + id="test_oldest_run_exactly_at_the_limit_then_continue", + ), + pytest.param("2024-05-29T23:59:59Z", None, id="test_oldest_run_past_the_limit_then_stop"), + pytest.param(None, "page-2", id="test_oldest_run_without_created_at_then_continue"), + ], +) +def test_stop_condition_can_compare_the_page_against_the_stream_slice( + oldest_created_at, expected_token +): + strategy = CursorPaginationStrategy( + cursor_value="{{ response.next }}", + stop_condition=( + "{{ str_to_datetime((response.get('runs') or [{}])[-1].get('created_at') or '9999-01-01T00:00:00Z')" + " < str_to_datetime(stream_interval['start_time']) - duration('P32D') }}" + ), + config={}, + parameters={}, + ) + response = requests.Response() + response._content = json.dumps( + {"next": "page-2", "runs": [{"created_at": oldest_created_at}]} + ).encode("utf-8") + stream_slice = StreamSlice( + partition={"repository": "airbytehq/airbyte"}, + cursor_slice={"start_time": "2024-07-01T00:00:00Z", "end_time": "2024-07-31T00:00:00Z"}, + ) + + assert ( + strategy.next_page_token(response, 1, None, None, stream_slice=stream_slice) + == expected_token + ) + + +def test_cursor_value_can_read_the_stream_partition(): + strategy = CursorPaginationStrategy( + cursor_value="{{ response.next ~ ':' ~ stream_partition['repository'] }}", + config={}, + parameters={}, + ) + response = requests.Response() + response._content = json.dumps({"next": "page-2"}).encode("utf-8") + stream_slice = StreamSlice(partition={"repository": "airbytehq/airbyte"}, cursor_slice={}) + + assert ( + strategy.next_page_token(response, 1, None, None, stream_slice=stream_slice) + == "page-2:airbytehq/airbyte" + ) diff --git a/unit_tests/sources/declarative/requesters/paginators/test_default_paginator.py b/unit_tests/sources/declarative/requesters/paginators/test_default_paginator.py index 5254e69d3d..eb11dc091c 100644 --- a/unit_tests/sources/declarative/requesters/paginators/test_default_paginator.py +++ b/unit_tests/sources/declarative/requesters/paginators/test_default_paginator.py @@ -26,6 +26,9 @@ from airbyte_cdk.sources.declarative.requesters.paginators.strategies.page_increment import ( PageIncrement, ) +from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import ( + PaginationStrategy, +) from airbyte_cdk.sources.declarative.requesters.request_path import RequestPath from airbyte_cdk.sources.declarative.types import Record, StreamSlice, StreamState @@ -674,3 +677,49 @@ def test_test_read_decorator_delegates_get_page_size(): paginator = PaginatorTestReadDecorator(decorated, 5) assert paginator.get_page_size() == 100 + + +class _StrategyWithoutStreamSlice(PaginationStrategy): + """A strategy written before `stream_slice` was passed to `next_page_token`, as custom ones in connectors are.""" + + @property + def initial_token(self): + return None + + def next_page_token(self, response, last_page_size, last_record, last_page_token_value=None): + return "next" + + def get_page_size(self): + return None + + +def test_given_a_strategy_accepting_stream_slice_when_next_page_token_then_forward_the_slice(): + strategy = MagicMock(spec=CursorPaginationStrategy) + strategy.next_page_token.return_value = "next" + paginator = DefaultPaginator( + pagination_strategy=strategy, config={}, url_base="https://airbyte.io", parameters={} + ) + response = requests.Response() + stream_slice = StreamSlice(partition={"repository": "a/b"}, cursor_slice={}) + + token = paginator.next_page_token(response, 2, None, None, stream_slice=stream_slice) + + assert token == {"next_page_token": "next"} + assert strategy.next_page_token.call_args.kwargs["stream_slice"] == stream_slice + + +@pytest.mark.parametrize("decorated", [False, True], ids=["test_default", "test_test_read"]) +def test_given_a_strategy_without_stream_slice_when_next_page_token_then_do_not_pass_it(decorated): + paginator = DefaultPaginator( + pagination_strategy=_StrategyWithoutStreamSlice(), + config={}, + url_base="https://airbyte.io", + parameters={}, + ) + if decorated: + paginator = PaginatorTestReadDecorator(paginator, maximum_number_of_pages=5) + stream_slice = StreamSlice(partition={"repository": "a/b"}, cursor_slice={}) + + token = paginator.next_page_token(requests.Response(), 2, None, None, stream_slice=stream_slice) + + assert token == {"next_page_token": "next"} diff --git a/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py b/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py index cb2ccc5987..f71e57accf 100644 --- a/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py +++ b/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py @@ -15,7 +15,7 @@ StopConditionPaginationStrategyDecorator, ) from airbyte_cdk.sources.streams.concurrent.cursor import Cursor -from airbyte_cdk.sources.types import Record +from airbyte_cdk.sources.types import Record, StreamSlice ANY_RECORD = Mock() NO_RECORD = None @@ -131,3 +131,19 @@ def test_given_page_size_override_when_next_page_token_then_forward_to_delegate( mocked_pagination_strategy.next_page_token.assert_called_once_with( ANY_RESPONSE, 25, NO_RECORD, None, page_size_override=25 ) + + +def test_given_stream_slice_when_next_page_token_then_forward_to_delegate( + mocked_pagination_strategy, mocked_stop_condition +): + mocked_stop_condition.is_met.return_value = False + decorator = StopConditionPaginationStrategyDecorator( + mocked_pagination_strategy, mocked_stop_condition + ) + stream_slice = StreamSlice(partition={"repository": "a/b"}, cursor_slice={}) + + decorator.next_page_token(ANY_RESPONSE, 25, NO_RECORD, None, stream_slice=stream_slice) + + mocked_pagination_strategy.next_page_token.assert_called_once_with( + ANY_RESPONSE, 25, NO_RECORD, None, stream_slice=stream_slice + ) diff --git a/unit_tests/sources/declarative/retrievers/test_simple_retriever.py b/unit_tests/sources/declarative/retrievers/test_simple_retriever.py index 1afa3b5fc9..c0fa483d78 100644 --- a/unit_tests/sources/declarative/retrievers/test_simple_retriever.py +++ b/unit_tests/sources/declarative/retrievers/test_simple_retriever.py @@ -1442,6 +1442,62 @@ def test_given_reach_pagination_limit_after_two_pages_when_read_records_than_red } +def test_given_pagination_reset_when_read_records_then_paginator_sees_the_slice_each_page_was_read_for(): + requester = Mock(spec=Requester) + requester.send_request.side_effect = [[{"id": 1}], [{"id": 2}], [{"id": 3}]] + record_selector = Mock(spec=HttpSelector) + record_selector.select_records.side_effect = [[{"id": 1}], [{"id": 2}], [{"id": 3}]] + pagination_tracker = Mock(spec=PaginationTracker) + pagination_tracker.has_reached_limit.side_effect = [False, True, False] + paginator = _mock_paginator() + paginator.get_initial_token.return_value = 1 + paginator.next_page_token.side_effect = [{"next_page_token": 2}, None] + retriever = SimpleRetriever( + name=A_STREAM_NAME, + primary_key=primary_key, + requester=requester, + record_selector=record_selector, + paginator=paginator, + pagination_tracker_factory=lambda: pagination_tracker, + parameters={}, + config={}, + ) + + list(retriever.read_records(A_RECORD_SCHEMA, A_STREAM_SLICE)) + + assert [call.kwargs["stream_slice"] for call in paginator.next_page_token.call_args_list] == [ + A_STREAM_SLICE, + pagination_tracker.reduce_slice_range_if_possible.return_value, + ] + + +class _PaginatorWithoutStreamSlice(DefaultPaginator): + """A paginator whose `next_page_token` predates the `stream_slice` argument.""" + + def next_page_token(self, response, last_page_size, last_record, last_page_token_value=None): + return None + + +def test_given_a_paginator_without_stream_slice_when_read_records_then_do_not_pass_it(): + requester = Mock(spec=Requester) + requester.send_request.return_value = [{"id": 1}] + record_selector = Mock(spec=HttpSelector) + record_selector.select_records.return_value = [{"id": 1}] + retriever = SimpleRetriever( + name=A_STREAM_NAME, + primary_key=primary_key, + requester=requester, + record_selector=record_selector, + paginator=_PaginatorWithoutStreamSlice( + pagination_strategy=Mock(), config={}, url_base="https://airbyte.io", parameters={} + ), + parameters={}, + config={}, + ) + + assert len(list(retriever.read_records(A_RECORD_SCHEMA, A_STREAM_SLICE))) == 1 + + @pytest.fixture(autouse=True) def _no_page_size_reduction_backoff(monkeypatch): """The reducer waits between reduction retries; taking those waits for real adds seconds to every CI run."""