allow for specifying PropertiesFromEndpoint from the HttpRequester to allow for interpolation on other request options

Author: brianjlai

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1459,7 +1459,7 @@ definitions:
         properties:
           type:
             type: string
-            enum: [ FileUploader ]
+            enum: [FileUploader]
           requester:
             description: Requester component that describes how to prepare HTTP requests to send to the source API.
             anyOf:
@@ -1960,6 +1960,10 @@ definitions:
           - "$ref": "#/definitions/DefaultErrorHandler"
           - "$ref": "#/definitions/CustomErrorHandler"
           - "$ref": "#/definitions/CompositeErrorHandler"
+      fetch_properties_from_endpoint:
+        title: Fetch Properties from Endpoint
+        description: Allows for retrieving a dynamic set of properties from an API endpoint which can be injected into outbound request using the stream_partition.extra_fields.
+        "$ref": "#/definitions/PropertiesFromEndpoint"
       http_method:
         title: HTTP Method
         description: The HTTP method used to fetch data from the source (can be GET or POST).
@@ -2351,7 +2355,7 @@ definitions:
     properties:
       type:
         type: string
-        enum: [ KeyTransformation ]
+        enum: [KeyTransformation]
       prefix:
         title: Key Prefix
         description: Prefix to add for object keys. If not provided original keys remain unchanged.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -880,20 +880,17 @@ class FlattenFields(BaseModel):
 
 
 class KeyTransformation(BaseModel):
-    prefix: Optional[Union[str, None]] = Field(
+    type: Literal["KeyTransformation"]
+    prefix: Optional[str] = Field(
         None,
         description="Prefix to add for object keys. If not provided original keys remain unchanged.",
-        examples=[
-            "flattened_",
-        ],
+        examples=["flattened_"],
         title="Key Prefix",
     )
-    suffix: Optional[Union[str, None]] = Field(
+    suffix: Optional[str] = Field(
         None,
         description="Suffix to add for object keys. If not provided original keys remain unchanged.",
-        examples=[
-            "_flattened",
-        ],
+        examples=["_flattened"],
         title="Key Suffix",
     )
 
@@ -916,7 +913,7 @@ class DpathFlattenFields(BaseModel):
         description="Whether to replace the origin record or not. Default is False.",
         title="Replace Origin Record",
     )
-    key_transformation: Optional[Union[KeyTransformation, None]] = Field(
+    key_transformation: Optional[KeyTransformation] = Field(
         None,
         description="Transformation for object keys. If not provided, original key will be used.",
         title="Key transformation",
@@ -2088,7 +2085,6 @@ class FileUploader(BaseModel):
             "{{ record.id }}_{{ record.file_name }}/",
         ],
     )
-    parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
 class DeclarativeStream(BaseModel):
@@ -2249,6 +2245,11 @@ class HttpRequester(BaseModel):
         description="Error handler component that defines how to handle errors.",
         title="Error Handler",
     )
+    fetch_properties_from_endpoint: Optional[PropertiesFromEndpoint] = Field(
+        None,
+        description="Allows for retrieving a dynamic set of properties from an API endpoint which can be injected into outbound request using the stream_partition.extra_fields.",
+        title="Fetch Properties from Endpoint",
+    )
     http_method: Optional[HttpMethod] = Field(
         HttpMethod.GET,
         description="The HTTP method used to fetch data from the source (can be GET or POST).",
@@ -2650,6 +2651,7 @@ class DynamicDeclarativeStream(BaseModel):
 FileUploader.update_forward_refs()
 DeclarativeStream.update_forward_refs()
 SessionTokenAuthenticator.update_forward_refs()
+HttpRequester.update_forward_refs()
 DynamicSchemaLoader.update_forward_refs()
 ParentStreamConfig.update_forward_refs()
 PropertiesFromEndpoint.update_forward_refs()

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2967,6 +2967,18 @@ def create_simple_retriever(
                 model.requester.request_parameters = self._remove_query_properties(
                     model.requester.request_parameters
                 )
+        elif model.requester.fetch_properties_from_endpoint:
+            query_properties_definition = QueryPropertiesModel(
+                type="QueryProperties",
+                property_list=model.requester.fetch_properties_from_endpoint,
+                always_include_properties=None,
+                property_chunking=None,
+            )  # type: ignore # $parameters has a default value
+
+            query_properties = self.create_query_properties(
+                model=query_properties_definition,
+                config=config,
+            )
 
         requester = self._create_component_from_model(
             model=model.requester,

airbyte_cdk/sources/declarative/requesters/query_properties/query_properties.py

@@ -46,13 +46,3 @@ def get_request_property_chunks(
             )
         else:
             yield list(fields)
-
-    # delete later, but leaving this to keep the discussion thread on the PR from getting hidden
-    def has_multiple_chunks(self, stream_slice: Optional[StreamSlice]) -> bool:
-        property_chunks = iter(self.get_request_property_chunks(stream_slice=stream_slice))
-        try:
-            next(property_chunks)
-            next(property_chunks)
-            return True
-        except StopIteration:
-            return False

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -4119,7 +4119,7 @@ def test_simple_retriever_with_query_properties():
     assert request_options_provider.request_parameters.get("nonary") == "{{config['nonary'] }}"
 
 
-def test_simple_retriever_with_properties_from_endpoint():
+def test_simple_retriever_with_request_parameters_properties_from_endpoint():
     content = """
     selector:
       type: RecordSelector
@@ -4216,6 +4216,88 @@ def test_simple_retriever_with_properties_from_endpoint():
     assert property_chunking.property_limit == 3
 
 
+def test_simple_retriever_with_requester_properties_from_endpoint():
+    content = """
+    selector:
+      type: RecordSelector
+      extractor:
+          type: DpathExtractor
+          field_path: ["extractor_path"]
+      record_filter:
+        type: RecordFilter
+        condition: "{{ record['id'] > stream_state['id'] }}"
+    requester:
+      type: HttpRequester
+      name: "{{ parameters['name'] }}"
+      url_base: "https://api.hubapi.com"
+      http_method: "GET"
+      path: "adAnalytics"
+      fetch_properties_from_endpoint:
+        type: PropertiesFromEndpoint
+        property_field_path: [ "name" ]
+        retriever:
+          type: SimpleRetriever
+          requester:
+            type: HttpRequester
+            url_base: https://api.hubapi.com
+            path: "/properties/v2/dynamics/properties"
+            http_method: GET
+          record_selector:
+            type: RecordSelector
+            extractor:
+              type: DpathExtractor
+              field_path: []
+    dynamic_properties_stream:
+      type: DeclarativeStream
+      incremental_sync:
+        type: DatetimeBasedCursor
+        $parameters:
+          datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z"
+        start_datetime: "{{ config['start_time'] }}"
+        cursor_field: "created"
+      retriever:
+        type: SimpleRetriever
+        name: "{{ parameters['name'] }}"
+        requester:
+          $ref: "#/requester"
+        record_selector:
+          $ref: "#/selector"
+      $parameters:
+        name: "dynamics"
+        """
+
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    stream_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["dynamic_properties_stream"], {}
+    )
+
+    stream = factory.create_component(
+        model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config
+    )
+
+    query_properties = stream.retriever.additional_query_properties
+    assert isinstance(query_properties, QueryProperties)
+    assert query_properties.always_include_properties is None
+    assert query_properties.property_chunking is None
+
+    properties_from_endpoint = stream.retriever.additional_query_properties.property_list
+    assert isinstance(properties_from_endpoint, PropertiesFromEndpoint)
+    assert properties_from_endpoint.property_field_path == ["name"]
+
+    properties_from_endpoint_retriever = (
+        stream.retriever.additional_query_properties.property_list.retriever
+    )
+    assert isinstance(properties_from_endpoint_retriever, SimpleRetriever)
+
+    properties_from_endpoint_requester = (
+        stream.retriever.additional_query_properties.property_list.retriever.requester
+    )
+    assert isinstance(properties_from_endpoint_requester, HttpRequester)
+    assert properties_from_endpoint_requester.url_base == "https://api.hubapi.com"
+    assert properties_from_endpoint_requester.path == "/properties/v2/dynamics/properties"
+
+
 def test_request_parameters_raise_error_if_not_of_type_query_properties():
     content = """
     selector: