Low code CDK: Allow nested objects for request_body_json (#26474)

* allow nested JSON

* add test for boolean

* review comment

* change for testing

* try fix

* try another fix

* Revert "change for testing"

This reverts commit 931b935.

* Revert "try fix"

This reverts commit 6f1c6c0.

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

@@ -1127,12 +1127,11 @@ definitions:
               }, "orderBy": 1, "columnName": "Timestamp"}]/
       request_body_json:
         title: Request Body JSON Payload
-        description: Specifies how to populate the body of the request with a JSON payload.
+        description: Specifies how to populate the body of the request with a JSON payload. Can contain nested objects.
         anyOf:
           - type: string
           - type: object
-            additionalProperties:
-              type: string
+            additionalProperties: true
         interpolation_context:
           - next_page_token
           - stream_interval
@@ -1143,6 +1142,9 @@ definitions:
           - sort_order: "ASC"
             sort_field: "CREATED_AT"
           - key: "{{ config['value'] }}"
+          - sort:
+              field: "updated_at"
+              order: "ascending"
       request_headers:
         title: Request Headers
         description: Return any non-auth headers. Authentication headers will overwrite any overlapping headers returned from this method.

airbyte-cdk/python/airbyte_cdk/sources/declarative/interpolation/interpolated_nested_mapping.py

@@ -0,0 +1,45 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+
+from dataclasses import InitVar, dataclass
+from typing import Any, Mapping, Optional, Union
+
+from airbyte_cdk.sources.declarative.interpolation.jinja import JinjaInterpolation
+from airbyte_cdk.sources.declarative.types import Config
+
+NestedMappingEntry = Union[dict[str, "NestedMapping"], list["NestedMapping"], str, int, float, bool, None]
+NestedMapping = Union[dict[str, NestedMappingEntry], str]
+
+
+@dataclass
+class InterpolatedNestedMapping:
+    """
+    Wrapper around a nested dict which can contain lists and primitive values where both the keys and values are interpolated recursively.
+
+    Attributes:
+        mapping (NestedMapping): to be evaluated
+    """
+
+    mapping: NestedMapping
+    parameters: InitVar[Mapping[str, Any]]
+
+    def __post_init__(self, parameters: Optional[Mapping[str, Any]]):
+        self._interpolation = JinjaInterpolation()
+        self._parameters = parameters
+
+    def eval(self, config: Config, **additional_parameters):
+        return self._eval(self.mapping, config, **additional_parameters)
+
+    def _eval(self, value, config, **kwargs):
+        # Recursively interpolate dictionaries and lists
+        if isinstance(value, str):
+            return self._interpolation.eval(value, config, parameters=self._parameters, **kwargs)
+        elif isinstance(value, dict):
+            interpolated_dict = {self._eval(k, config, **kwargs): self._eval(v, config, **kwargs) for k, v in value.items()}
+            return {k: v for k, v in interpolated_dict.items() if v is not None}
+        elif isinstance(value, list):
+            return [self._eval(v, config, **kwargs) for v in value]
+        else:
+            return value

airbyte-cdk/python/airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1094,12 +1094,13 @@ class HttpRequester(BaseModel):
         ],
         title="Request Body Payload (Non-JSON)",
     )
-    request_body_json: Optional[Union[str, Dict[str, str]]] = Field(
+    request_body_json: Optional[Union[str, Dict[str, Any]]] = Field(
         None,
-        description="Specifies how to populate the body of the request with a JSON payload.",
+        description="Specifies how to populate the body of the request with a JSON payload. Can contain nested objects.",
         examples=[
             {"sort_order": "ASC", "sort_field": "CREATED_AT"},
             {"key": "{{ config['value'] }}"},
+            {"sort": {"field": "updated_at", "order": "ascending"}},
         ],
         title="Request Body JSON Payload",
     )

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/request_options/interpolated_nested_request_input_provider.py

@@ -0,0 +1,45 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import InitVar, dataclass, field
+from typing import Any, Mapping, Optional, Union
+
+from airbyte_cdk.sources.declarative.interpolation.interpolated_nested_mapping import InterpolatedNestedMapping, NestedMapping
+from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
+from airbyte_cdk.sources.declarative.types import Config, StreamSlice, StreamState
+
+
+@dataclass
+class InterpolatedNestedRequestInputProvider:
+    """
+    Helper class that generically performs string interpolation on a provided deeply nested dictionary or string input
+    """
+
+    parameters: InitVar[Mapping[str, Any]]
+    request_inputs: Optional[NestedMapping] = field(default=None)
+    config: Config = field(default_factory=dict)
+    _interpolator: Union[InterpolatedString, InterpolatedNestedMapping] = field(init=False, repr=False, default=None)
+    _request_inputs: Union[str, Mapping[str, str]] = field(init=False, repr=False, default=None)
+
+    def __post_init__(self, parameters: Mapping[str, Any]):
+
+        self._request_inputs = self.request_inputs or {}
+        if isinstance(self.request_inputs, str):
+            self._interpolator = InterpolatedString(self.request_inputs, default="", parameters=parameters)
+        else:
+            self._interpolator = InterpolatedNestedMapping(self._request_inputs, parameters=parameters)
+
+    def eval_request_inputs(
+        self, stream_state: StreamState, stream_slice: Optional[StreamSlice] = None, next_page_token: Mapping[str, Any] = None
+    ) -> Mapping[str, Any]:
+        """
+        Returns the request inputs to set on an outgoing HTTP request
+
+        :param stream_state: The stream state
+        :param stream_slice: The stream slice
+        :param next_page_token: The pagination token
+        :return: The request inputs to set on an outgoing HTTP request
+        """
+        kwargs = {"stream_state": stream_state, "stream_slice": stream_slice, "next_page_token": next_page_token}
+        return self._interpolator.eval(self.config, **kwargs)

airbyte-cdk/python/airbyte_cdk/sources/declarative/requesters/request_options/interpolated_request_options_provider.py

@@ -5,6 +5,10 @@
 from dataclasses import InitVar, dataclass, field
 from typing import Any, Mapping, MutableMapping, Optional, Union
 
+from airbyte_cdk.sources.declarative.interpolation.interpolated_nested_mapping import NestedMapping
+from airbyte_cdk.sources.declarative.requesters.request_options.interpolated_nested_request_input_provider import (
+    InterpolatedNestedRequestInputProvider,
+)
 from airbyte_cdk.sources.declarative.requesters.request_options.interpolated_request_input_provider import InterpolatedRequestInputProvider
 from airbyte_cdk.sources.declarative.requesters.request_options.request_options_provider import RequestOptionsProvider
 from airbyte_cdk.sources.declarative.types import Config, StreamSlice, StreamState
@@ -30,7 +34,7 @@ class InterpolatedRequestOptionsProvider(RequestOptionsProvider):
     request_parameters: Optional[RequestInput] = None
     request_headers: Optional[RequestInput] = None
     request_body_data: Optional[RequestInput] = None
-    request_body_json: Optional[RequestInput] = None
+    request_body_json: Optional[NestedMapping] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]):
         if self.request_parameters is None:
@@ -54,7 +58,7 @@ def __post_init__(self, parameters: Mapping[str, Any]):
         self._body_data_interpolator = InterpolatedRequestInputProvider(
             config=self.config, request_inputs=self.request_body_data, parameters=parameters
         )
-        self._body_json_interpolator = InterpolatedRequestInputProvider(
+        self._body_json_interpolator = InterpolatedNestedRequestInputProvider(
             config=self.config, request_inputs=self.request_body_json, parameters=parameters
         )
 

airbyte-cdk/python/unit_tests/sources/declarative/interpolation/test_interpolated_nested_mapping.py

@@ -0,0 +1,45 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+import dpath.util
+import pytest
+from airbyte_cdk.sources.declarative.interpolation.interpolated_nested_mapping import InterpolatedNestedMapping
+
+
+@pytest.mark.parametrize(
+    "test_name, path, expected_value",
+    [
+        ("test_field_value", "nested/field", "value"),
+        ("test_number", "nested/number", 100),
+        ("test_interpolated_number", "nested/nested_array/1/value", 5),
+        ("test_interpolated_boolean", "nested/nested_array/2/value", True),
+        ("test_field_to_interpolate_from_config", "nested/config_value", "VALUE_FROM_CONFIG"),
+        ("test_field_to_interpolate_from_kwargs", "nested/kwargs_value", "VALUE_FROM_KWARGS"),
+        ("test_field_to_interpolate_from_parameters", "nested/parameters_value", "VALUE_FROM_PARAMETERS"),
+        ("test_key_is_interpolated", "nested/nested_array/0/key", "VALUE"),
+    ],
+)
+def test(test_name, path, expected_value):
+    d = {
+            "nested": {
+                "field": "value",
+                "number": 100,
+                "nested_array": [
+                    {"{{ parameters.k }}": "VALUE"},
+                    {"value": "{{ config['num_value'] | int + 2 }}"},
+                    {"value": "{{ True }}"},
+                ],
+                "config_value": "{{ config['c'] }}",
+                "parameters_value": "{{ parameters['b'] }}",
+                "kwargs_value": "{{ kwargs['a'] }}",
+            }
+        }
+
+    config = {"c": "VALUE_FROM_CONFIG", "num_value": 3}
+    kwargs = {"a": "VALUE_FROM_KWARGS"}
+    mapping = InterpolatedNestedMapping(mapping=d, parameters={"b": "VALUE_FROM_PARAMETERS", "k": "key"})
+
+    interpolated = mapping.eval(config, **{"kwargs": kwargs})
+
+    assert dpath.util.get(interpolated, path) == expected_value

airbyte-cdk/python/unit_tests/sources/declarative/requesters/request_options/test_interpolated_request_options_provider.py

@@ -59,6 +59,9 @@ def test_interpolated_request_params(test_name, input_request_params, expected_r
         ("test_number_falsy_value", {"number_falsy": "{{ 0.0 }}"}, {"number_falsy": 0.0}),
         ("test_string_falsy_value", {"string_falsy": "{{ '' }}"}, {}),
         ("test_none_value", {"none_value": "{{ None }}"}, {}),
+        ("test_string", """{"nested": { "key": "{{ config['option'] }}" }}""", {"nested": {"key": "OPTION"}}),
+        ("test_nested_objects", {"nested": {"key": "{{ config['option'] }}"}}, {"nested": {"key": "OPTION"}}),
+        ("test_nested_objects_interpolated keys", {"nested": {"{{ stream_state['date'] }}": "{{ config['option'] }}"}}, {"nested": {"2021-01-01": "OPTION"}}),
     ],
 )
 def test_interpolated_request_json(test_name, input_request_json, expected_request_json):