feat(declarative): Add interpolation support for CursorPagination page_size (#878)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Alfredo Garcia <alfredo.garcia@hallmark.edu>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -512,9 +512,16 @@ definitions:
       page_size:
         title: Page Size
         description: The number of records to include in each pages.
-        type: integer
+        anyOf:
+          - type: integer
+            title: Number of Records
+          - type: string
+            title: Interpolated Value
+        interpolation_context:
+          - config
         examples:
           - 100
+          - "{{ config['page_size'] }}"
       stop_condition:
         title: Stop Condition
         description: Template string evaluating when to stop paginating.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -114,10 +114,10 @@ class CursorPagination(BaseModel):
         ],
         title="Cursor Value",
     )
-    page_size: Optional[int] = Field(
+    page_size: Optional[Union[int, str]] = Field(
         None,
         description="The number of records to include in each pages.",
-        examples=[100],
+        examples=[100, "{{ config['page_size'] }}"],
         title="Page Size",
     )
     stop_condition: Optional[str] = Field(
@@ -2741,7 +2741,7 @@ class HttpRequester(BaseModelWithDeprecations):
     )
     use_cache: Optional[bool] = Field(
         False,
-        description="Enables stream requests caching. This field is automatically set by the CDK.",
+        description="Enables stream requests caching. When set to true, repeated requests to the same URL will return cached responses. Parent streams automatically have caching enabled. Only set this to false if you are certain that caching should be disabled, as it may negatively impact performance when the same data is needed multiple times (e.g., for scroll-based pagination APIs where caching causes duplicate records).",
         title="Use Cache",
     )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -1554,14 +1554,18 @@ def create_concurrent_cursor_from_incrementing_count_cursor(
                 f"Expected {model_type.__name__} component, but received {incrementing_count_cursor_model.__class__.__name__}"
             )
 
-        interpolated_start_value = (
-            InterpolatedString.create(
-                incrementing_count_cursor_model.start_value,  # type: ignore
+        start_value: Union[int, str, None] = incrementing_count_cursor_model.start_value
+        # Pydantic Union type coercion can convert int 0 to string '0' depending on Union order.
+        # We need to handle both int and str representations of numeric values.
+        # Evaluate the InterpolatedString and convert to int for the ConcurrentCursor.
+        if start_value is not None:
+            interpolated_start_value = InterpolatedString.create(
+                str(start_value),  # Ensure we pass a string to InterpolatedString.create
                 parameters=incrementing_count_cursor_model.parameters or {},
             )
-            if incrementing_count_cursor_model.start_value
-            else 0
-        )
+            evaluated_start_value: int = int(interpolated_start_value.eval(config=config))
+        else:
+            evaluated_start_value = 0
 
         cursor_field = self._get_catalog_defined_cursor_field(
             stream_name=stream_name,
@@ -1593,7 +1597,7 @@ def create_concurrent_cursor_from_incrementing_count_cursor(
             connector_state_converter=connector_state_converter,
             cursor_field=cursor_field,
             slice_boundary_fields=None,
-            start=interpolated_start_value,  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
+            start=evaluated_start_value,  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
             end_provider=connector_state_converter.get_end_provider(),  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
         )
 
@@ -1745,10 +1749,16 @@ def create_cursor_pagination(
                 self._UNSUPPORTED_DECODER_ERROR.format(decoder_type=type(inner_decoder))
             )
 
+        # 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
+        if isinstance(page_size, str) and page_size.isdigit():
+            page_size = int(page_size)
+
         return CursorPaginationStrategy(
             cursor_value=model.cursor_value,
             decoder=decoder_to_use,
-            page_size=model.page_size,
+            page_size=page_size,
             stop_condition=model.stop_condition,
             config=config,
             parameters=model.parameters or {},
@@ -2917,8 +2927,14 @@ def create_offset_increment(
             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
+        if isinstance(page_size, str) and page_size.isdigit():
+            page_size = int(page_size)
+
         return OffsetIncrement(
-            page_size=model.page_size,
+            page_size=page_size,
             config=config,
             decoder=decoder_to_use,
             extractor=extractor,
@@ -2930,8 +2946,14 @@ def create_offset_increment(
     def create_page_increment(
         model: PageIncrementModel, config: Config, **kwargs: Any
     ) -> PageIncrement:
+        # 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
+        if isinstance(page_size, str) and page_size.isdigit():
+            page_size = int(page_size)
+
         return PageIncrement(
-            page_size=model.page_size,
+            page_size=page_size,
             config=config,
             start_from_page=model.start_from_page or 0,
             inject_on_first_request=model.inject_on_first_request or False,

airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py

@@ -23,10 +23,10 @@
 @dataclass
 class CursorPaginationStrategy(PaginationStrategy):
     """
-    Pagination strategy that evaluates an interpolated string to define the next page token
+    Pagination strategy that evaluates an interpolated string to define the next page token.
 
     Attributes:
-        page_size (Optional[int]): the number of records to request
+        page_size (Optional[Union[str, int]]): the number of records to request
         cursor_value (Union[InterpolatedString, str]): template string evaluating to the cursor value
         config (Config): connection config
         stop_condition (Optional[InterpolatedBoolean]): template string evaluating when to stop paginating
@@ -36,7 +36,7 @@ class CursorPaginationStrategy(PaginationStrategy):
     cursor_value: Union[InterpolatedString, str]
     config: Config
     parameters: InitVar[Mapping[str, Any]]
-    page_size: Optional[int] = None
+    page_size: Optional[Union[str, int]] = None
     stop_condition: Optional[Union[InterpolatedBoolean, str]] = None
     decoder: Decoder = field(
         default_factory=lambda: PaginationDecoderDecorator(decoder=JsonDecoder(parameters={}))
@@ -54,6 +54,14 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         else:
             self._stop_condition = self.stop_condition
 
+        if isinstance(self.page_size, int) or (self.page_size is None):
+            self._page_size = self.page_size
+        else:
+            page_size = InterpolatedString(self.page_size, parameters=parameters).eval(self.config)
+            if not isinstance(page_size, int):
+                raise Exception(f"{page_size} is of type {type(page_size)}. Expected {int}")
+            self._page_size = page_size
+
     @property
     def initial_token(self) -> Optional[Any]:
         """
@@ -95,4 +103,4 @@ def next_page_token(
         return token if token else None
 
     def get_page_size(self) -> Optional[int]:
-        return self.page_size
+        return self._page_size

unit_tests/sources/declarative/requesters/paginators/test_cursor_pagination_strategy.py

@@ -112,3 +112,43 @@ def test_last_record_is_node_if_no_records():
     response = requests.Response()
     next_page_token = strategy.next_page_token(response, 0, None)
     assert next_page_token is None
+
+
+@pytest.mark.parametrize(
+    "page_size_input, config, expected_page_size",
+    [
+        pytest.param(100, {}, 100, id="static_integer"),
+        pytest.param("100", {}, 100, id="static_string"),
+        pytest.param(
+            "{{ config['page_size'] }}", {"page_size": 50}, 50, id="interpolated_from_config"
+        ),
+        pytest.param("{{ config.get('page_size', 100) }}", {}, 100, id="interpolated_with_default"),
+        pytest.param(
+            "{{ config.get('page_size', 100) }}",
+            {"page_size": 200},
+            200,
+            id="interpolated_override_default",
+        ),
+        pytest.param(None, {}, None, id="none_page_size"),
+    ],
+)
+def test_interpolated_page_size(page_size_input, config, expected_page_size):
+    """Test that page_size supports interpolation from config."""
+    strategy = CursorPaginationStrategy(
+        page_size=page_size_input,
+        cursor_value="token",
+        config=config,
+        parameters={},
+    )
+    assert strategy.get_page_size() == expected_page_size
+
+
+def test_interpolated_page_size_raises_on_non_integer():
+    """Test that initialization raises an exception when interpolation resolves to a non-integer."""
+    with pytest.raises(Exception, match="is of type .* Expected"):
+        CursorPaginationStrategy(
+            page_size="{{ config['page_size'] }}",
+            cursor_value="token",
+            config={"page_size": "invalid"},
+            parameters={},
+        )