Airbyte CDK: add interpolation for request options

airbyte-cdk/python/airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1898,6 +1898,9 @@ definitions:
         type: string
         examples:
           - segment_id
+        interpolation_context:
+          - config
+          - parameters
       inject_into:
         title: Inject Into
         description: Configures where the descriptor should be set on the HTTP requests. Note that request parameters that are already encoded in the URL path will not be duplicated.

airbyte-cdk/python/airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py

@@ -247,9 +247,11 @@ def request_kwargs(self) -> Mapping[str, Any]:
     def _get_request_options(self, option_type: RequestOptionType, stream_slice: StreamSlice):
         options = {}
         if self.start_time_option and self.start_time_option.inject_into == option_type:
-            options[self.start_time_option.field_name] = stream_slice.get(self.partition_field_start.eval(self.config))
+            options[self.start_time_option.field_name.eval(config=self.config)] = stream_slice.get(
+                self.partition_field_start.eval(self.config)
+            )
         if self.end_time_option and self.end_time_option.inject_into == option_type:
-            options[self.end_time_option.field_name] = stream_slice.get(self.partition_field_end.eval(self.config))
+            options[self.end_time_option.field_name.eval(config=self.config)] = stream_slice.get(self.partition_field_end.eval(self.config))
         return options
 
     def should_be_synced(self, record: Record) -> bool:

airbyte-cdk/python/airbyte_cdk/sources/declarative/partition_routers/list_partition_router.py

@@ -81,7 +81,7 @@ def _get_request_option(self, request_option_type: RequestOptionType, stream_sli
         if self.request_option and self.request_option.inject_into == request_option_type and stream_slice:
             slice_value = stream_slice.get(self.cursor_field.eval(self.config))
             if slice_value:
-                return {self.request_option.field_name: slice_value}
+                return {self.request_option.field_name.eval(self.config): slice_value}
             else:
                 return {}
         else:

airbyte-cdk/python/airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py

@@ -100,7 +100,7 @@ def _get_request_option(self, option_type: RequestOptionType, stream_slice: Stre
                     key = parent_config.partition_field.eval(self.config)
                     value = stream_slice.get(key)
                     if value:
-                        params.update({parent_config.request_option.field_name: value})
+                        params.update({parent_config.request_option.field_name.eval(config=self.config): value})
         return params
 
     def stream_slices(self) -> Iterable[StreamSlice]:

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/paginators/default_paginator.py

@@ -164,9 +164,9 @@ def _get_request_options(self, option_type: RequestOptionType) -> MutableMapping
             and isinstance(self.page_token_option, RequestOption)
             and self.page_token_option.inject_into == option_type
         ):
-            options[self.page_token_option.field_name] = self._token
+            options[self.page_token_option.field_name.eval(config=self.config)] = self._token
         if self.page_size_option and self.pagination_strategy.get_page_size() and self.page_size_option.inject_into == option_type:
-            options[self.page_size_option.field_name] = self.pagination_strategy.get_page_size()
+            options[self.page_size_option.field_name.eval(config=self.config)] = self.pagination_strategy.get_page_size()
         return options
 
 

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/request_option.py

@@ -4,7 +4,9 @@
 
 from dataclasses import InitVar, dataclass
 from enum import Enum
-from typing import Any, Mapping
+from typing import Any, Mapping, Union
+
+from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 
 
 class RequestOptionType(Enum):
@@ -28,6 +30,9 @@ class RequestOption:
         inject_into (RequestOptionType): Describes where in the HTTP request to inject the parameter
     """
 
-    field_name: str
+    field_name: Union[InterpolatedString, str]
     inject_into: RequestOptionType
     parameters: InitVar[Mapping[str, Any]]
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        self.field_name = InterpolatedString.create(self.field_name, parameters=parameters)

airbyte-cdk/python/unit_tests/sources/declarative/auth/test_token_auth.py

@@ -83,7 +83,7 @@ def test_api_key_authenticator(test_name, header, token, expected_header, expect
     """
     token_provider = InterpolatedStringTokenProvider(config=config, api_token=token, parameters=parameters)
     token_auth = ApiKeyAuthenticator(
-        request_option=RequestOption(inject_into=RequestOptionType.header, field_name=header, parameters={}),
+        request_option=RequestOption(inject_into=RequestOptionType.header, field_name=header, parameters=parameters),
         token_provider=token_provider,
         config=config,
         parameters=parameters,
@@ -192,7 +192,7 @@ def test_api_key_authenticator_inject(test_name, field_name, token, expected_fie
     """
     token_provider = InterpolatedStringTokenProvider(config=config, api_token=token, parameters=parameters)
     token_auth = ApiKeyAuthenticator(
-        request_option=RequestOption(inject_into=inject_type, field_name=field_name, parameters={}),
+        request_option=RequestOption(inject_into=inject_type, field_name=field_name, parameters=parameters),
         token_provider=token_provider,
         config=config,
         parameters=parameters,

airbyte-cdk/python/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -232,7 +232,7 @@ def test_full_config_stream():
 
     assert isinstance(stream.retriever.paginator, DefaultPaginator)
     assert isinstance(stream.retriever.paginator.decoder, JsonDecoder)
-    assert stream.retriever.paginator.page_size_option.field_name == "page_size"
+    assert stream.retriever.paginator.page_size_option.field_name.eval(input_config) == "page_size"
     assert stream.retriever.paginator.page_size_option.inject_into == RequestOptionType.request_parameter
     assert isinstance(stream.retriever.paginator.page_token_option, RequestPath)
     assert stream.retriever.paginator.url_base.string == "https://api.sendgrid.com/v3/"
@@ -422,7 +422,7 @@ def test_list_based_stream_slicer_with_values_defined_in_config():
     assert isinstance(partition_router, ListPartitionRouter)
     assert partition_router.values == ["airbyte", "airbyte-cloud"]
     assert partition_router.request_option.inject_into == RequestOptionType.header
-    assert partition_router.request_option.field_name == "repository"
+    assert partition_router.request_option.field_name.eval(config=input_config) == "repository"
 
 
 def test_create_substream_partition_router():
@@ -484,7 +484,7 @@ def test_create_substream_partition_router():
     assert partition_router.parent_stream_configs[0].parent_key.eval({}) == "id"
     assert partition_router.parent_stream_configs[0].partition_field.eval({}) == "repository_id"
     assert partition_router.parent_stream_configs[0].request_option.inject_into == RequestOptionType.request_parameter
-    assert partition_router.parent_stream_configs[0].request_option.field_name == "repository_id"
+    assert partition_router.parent_stream_configs[0].request_option.field_name.eval(config=input_config) == "repository_id"
 
     assert partition_router.parent_stream_configs[1].parent_key.eval({}) == "someid"
     assert partition_router.parent_stream_configs[1].partition_field.eval({}) == "word_id"
@@ -529,9 +529,9 @@ def test_datetime_based_cursor():
     assert stream_slicer.cursor_granularity == "PT0.000001S"
     assert stream_slicer.lookback_window.string == "P5D"
     assert stream_slicer.start_time_option.inject_into == RequestOptionType.request_parameter
-    assert stream_slicer.start_time_option.field_name == "created[gte]"
+    assert stream_slicer.start_time_option.field_name.eval(config=input_config) == "created[gte]"
     assert stream_slicer.end_time_option.inject_into == RequestOptionType.body_json
-    assert stream_slicer.end_time_option.field_name == "end_time"
+    assert stream_slicer.end_time_option.field_name.eval({}) == "end_time"
     assert stream_slicer.partition_field_start.eval({}) == "star"
     assert stream_slicer.partition_field_end.eval({}) == "en"
 
@@ -1121,7 +1121,7 @@ def test_create_default_paginator():
 
     assert isinstance(paginator.page_size_option, RequestOption)
     assert paginator.page_size_option.inject_into == RequestOptionType.request_parameter
-    assert paginator.page_size_option.field_name == "page_size"
+    assert paginator.page_size_option.field_name.eval(config=input_config) == "page_size"
 
     assert isinstance(paginator.page_token_option, RequestPath)
 
@@ -1294,7 +1294,7 @@ def test_custom_components_do_not_contain_extra_fields():
     assert custom_substream_partition_router.parent_stream_configs[0].parent_key.eval({}) == "id"
     assert custom_substream_partition_router.parent_stream_configs[0].partition_field.eval({}) == "repository_id"
     assert custom_substream_partition_router.parent_stream_configs[0].request_option.inject_into == RequestOptionType.request_parameter
-    assert custom_substream_partition_router.parent_stream_configs[0].request_option.field_name == "repository_id"
+    assert custom_substream_partition_router.parent_stream_configs[0].request_option.field_name.eval(config=input_config) == "repository_id"
 
     assert isinstance(custom_substream_partition_router.custom_pagination_strategy, PageIncrement)
     assert custom_substream_partition_router.custom_pagination_strategy.page_size == 100
@@ -1343,7 +1343,7 @@ def test_parse_custom_component_fields_if_subcomponent():
     assert custom_substream_partition_router.parent_stream_configs[0].parent_key.eval({}) == "id"
     assert custom_substream_partition_router.parent_stream_configs[0].partition_field.eval({}) == "repository_id"
     assert custom_substream_partition_router.parent_stream_configs[0].request_option.inject_into == RequestOptionType.request_parameter
-    assert custom_substream_partition_router.parent_stream_configs[0].request_option.field_name == "repository_id"
+    assert custom_substream_partition_router.parent_stream_configs[0].request_option.field_name.eval(config=input_config) == "repository_id"
 
     assert isinstance(custom_substream_partition_router.custom_pagination_strategy, PageIncrement)
     assert custom_substream_partition_router.custom_pagination_strategy.page_size == 100

airbyte-cdk/python/unit_tests/sources/declarative/requesters/paginators/test_request_option.py

@@ -7,15 +7,19 @@
 
 
 @pytest.mark.parametrize(
-    "test_name, option_type, field_name",
+    "test_name, option_type, field_name, expected_field_name",
     [
-        ("test_limit_param_with_field_name", RequestOptionType.request_parameter, "field"),
-        ("test_limit_header_with_field_name", RequestOptionType.header, "field"),
-        ("test_limit_data_with_field_name", RequestOptionType.body_data, "field"),
-        ("test_limit_json_with_field_name", RequestOptionType.body_json, "field"),
+        ("test_limit_param_with_field_name", RequestOptionType.request_parameter, "field", "field"),
+        ("test_limit_header_with_field_name", RequestOptionType.header, "field", "field"),
+        ("test_limit_data_with_field_name", RequestOptionType.body_data, "field", "field"),
+        ("test_limit_json_with_field_name", RequestOptionType.body_json, "field", "field"),
+        ("test_limit_json_with_field_name", RequestOptionType.request_parameter, "since_{{ parameters['cursor_field'] }}", "since_updated_at"),
+        ("test_limit_header_with_field_name", RequestOptionType.header, "since_{{ parameters['cursor_field'] }}", "since_updated_at"),
+        ("test_limit_data_with_field_name", RequestOptionType.body_data, "since_{{ parameters['cursor_field'] }}", "since_updated_at"),
+        ("test_limit_json_with_field_name", RequestOptionType.body_json, "since_{{ parameters['cursor_field'] }}", "since_updated_at"),
     ],
 )
-def test_request_option(test_name, option_type, field_name):
-    request_option = RequestOption(inject_into=option_type, field_name=field_name, parameters={})
-    assert request_option.field_name == field_name
+def test_request_option(test_name, option_type, field_name: str, expected_field_name: str):
+    request_option = RequestOption(inject_into=option_type, field_name=field_name, parameters={"cursor_field": "updated_at"})
+    assert request_option.field_name.eval({}) == expected_field_name
     assert request_option.inject_into == option_type