From d3e2e209cdc791af9b32994b1245f67d31dd4c57 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 14:14:57 +0000 Subject: [PATCH] fix: use raw record count in PageIncrement so record filters do not stop pagination early Co-Authored-By: Daryna Ishchenko --- .../parsers/model_to_component_factory.py | 18 +++++- .../paginators/strategies/page_increment.py | 8 +++ .../paginators/test_page_increment.py | 55 +++++++++++++++++++ 3 files changed, 79 insertions(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 5a75912b36..c2551b4859 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -3032,10 +3032,23 @@ def create_offset_increment( parameters=model.parameters or {}, ) - @staticmethod def create_page_increment( - model: PageIncrementModel, config: Config, **kwargs: Any + self, + model: PageIncrementModel, + config: Config, + decoder: Optional[Decoder] = None, + extractor_model: Optional[Union[CustomRecordExtractorModel, DpathExtractorModel]] = None, + **kwargs: Any, ) -> PageIncrement: + # Like OffsetIncrement, we instantiate a separate extractor with identical behavior to the + # RecordSelector's so the strategy can count the raw records in the response. This ensures + # pagination is driven by the API's page size, not the post-filter record count. + extractor = ( + self._create_component_from_model(model=extractor_model, config=config, decoder=decoder) + if extractor_model + else None + ) + # Pydantic v1 Union type coercion can convert int to string depending on Union order. # If page_size is a string that represents an integer (not an interpolation), convert it back. page_size = model.page_size @@ -3047,6 +3060,7 @@ def create_page_increment( config=config, start_from_page=model.start_from_page or 0, inject_on_first_request=model.inject_on_first_request or False, + extractor=extractor, parameters=model.parameters or {}, ) 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 2e1643b565..f3bad7152b 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/page_increment.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/page_increment.py @@ -7,6 +7,7 @@ import requests +from airbyte_cdk.sources.declarative.extractors.record_extractor import RecordExtractor from airbyte_cdk.sources.declarative.interpolation import InterpolatedString from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import ( PaginationStrategy, @@ -29,6 +30,7 @@ class PageIncrement(PaginationStrategy): parameters: InitVar[Mapping[str, Any]] start_from_page: int = 0 inject_on_first_request: bool = False + extractor: Optional[RecordExtractor] = None def __post_init__(self, parameters: Mapping[str, Any]) -> None: if isinstance(self.page_size, int) or (self.page_size is None): @@ -52,6 +54,12 @@ def next_page_token( last_record: Optional[Record], last_page_token_value: Optional[Any], ) -> Optional[Any]: + if self.extractor: + # The record count is dependent on the records returned from the response which may not always + # align with the size of pages emitted. For example, a record filter can reduce the number of + # records observed below the page size even though the API returned a full page. + last_page_size = len(list(self.extractor.extract_records(response=response))) + # Stop paginating when there are fewer records than the page size or the current page has no records if (self._page_size and last_page_size < self._page_size) or last_page_size == 0: return None diff --git a/unit_tests/sources/declarative/requesters/paginators/test_page_increment.py b/unit_tests/sources/declarative/requesters/paginators/test_page_increment.py index 32af20b50d..ecb458b951 100644 --- a/unit_tests/sources/declarative/requesters/paginators/test_page_increment.py +++ b/unit_tests/sources/declarative/requesters/paginators/test_page_increment.py @@ -8,11 +8,66 @@ import pytest import requests +from airbyte_cdk.sources.declarative.extractors import DpathExtractor from airbyte_cdk.sources.declarative.requesters.paginators.strategies.page_increment import ( PageIncrement, ) +@pytest.mark.parametrize( + "response_results, last_page_size, last_page_token_value, expected_next_page_token", + [ + pytest.param( + [{"id": 1}, {"id": 2}], + 0, + 3, + 4, + id="test_full_page_continues_even_if_all_records_filtered", + ), + pytest.param( + [{"id": 1}, {"id": 2}], + 1, + 3, + 4, + id="test_full_page_continues_even_if_some_records_filtered", + ), + pytest.param( + [{"id": 1}], + 1, + 3, + None, + id="test_partial_page_stops_pagination", + ), + pytest.param( + [], + 0, + 3, + None, + id="test_empty_page_stops_pagination", + ), + ], +) +def test_page_increment_paginator_strategy_with_extractor( + response_results, last_page_size, last_page_token_value, expected_next_page_token +): + extractor = DpathExtractor(field_path=["results"], parameters={}, config={}) + paginator_strategy = PageIncrement( + page_size=2, parameters={}, start_from_page=1, extractor=extractor, config={} + ) + + response = requests.Response() + response.headers = {"A_HEADER": "HEADER_VALUE"} + response._content = json.dumps({"results": response_results}).encode("utf-8") + + next_page_token = paginator_strategy.next_page_token( + response, + last_page_size, + response_results[-1] if response_results else None, + last_page_token_value, + ) + assert expected_next_page_token == next_page_token + + @pytest.mark.parametrize( "page_size, start_from, last_page_size, last_record, last_page_token_value, expected_next_page_token, expected_offset", [