🐛 [low-code] $options shouldn't overwrite values that are already def…

…ined (#18060)

* fix

* Add missing test

* remove prints

* extract to method

* rename

* Add missing test

* rename

* bump

airbyte-cdk/python/CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 0.1.101
+
+- Low-code: $options do not overwrite parameters that are already set
+
 ## 0.1.100
 
 - Low-code: Pass stream_slice to read_records when reading from CheckStream

airbyte-cdk/python/airbyte_cdk/sources/declarative/create_partial.py

@@ -3,6 +3,7 @@
 #
 
 import inspect
+from typing import Any, Mapping
 
 OPTIONS_STR = "$options"
 
@@ -23,6 +24,7 @@ def create(func, /, *args, **keywords):
     """
 
     def newfunc(*fargs, **fkeywords):
+
         all_keywords = {**keywords}
         all_keywords.update(fkeywords)
 
@@ -39,8 +41,8 @@ def newfunc(*fargs, **fkeywords):
         if config is not None:
             all_keywords["config"] = config
 
-        kwargs_to_pass_down = _get_kwargs_to_pass_to_func(func, options)
-        all_keywords_to_pass_down = _get_kwargs_to_pass_to_func(func, all_keywords)
+        kwargs_to_pass_down = _get_kwargs_to_pass_to_func(func, options, all_keywords)
+        all_keywords_to_pass_down = _get_kwargs_to_pass_to_func(func, all_keywords, all_keywords)
 
         # options is required as part of creation of all declarative components
         dynamic_args = {**all_keywords_to_pass_down, **kwargs_to_pass_down}
@@ -63,17 +65,23 @@ def newfunc(*fargs, **fkeywords):
     return newfunc
 
 
-def _get_kwargs_to_pass_to_func(func, options):
+def _get_kwargs_to_pass_to_func(func, options, existing_keyword_parameters):
     argspec = inspect.getfullargspec(func)
     kwargs_to_pass_down = set(argspec.kwonlyargs)
     args_to_pass_down = set(argspec.args)
     all_args = args_to_pass_down.union(kwargs_to_pass_down)
-    kwargs_to_pass_down = {k: v for k, v in options.items() if k in all_args}
+    kwargs_to_pass_down = {
+        k: v for k, v in options.items() if k in all_args and _key_is_unset_or_identical(k, v, existing_keyword_parameters)
+    }
     if "options" in all_args:
         kwargs_to_pass_down["options"] = options
     return kwargs_to_pass_down
 
 
+def _key_is_unset_or_identical(key: str, value: Any, mapping: Mapping[str, Any]):
+    return key not in mapping or mapping[key] == value
+
+
 def _create_inner_objects(keywords, kwargs):
     fully_created = dict()
     for k, v in keywords.items():

airbyte-cdk/python/setup.py

@@ -15,7 +15,7 @@
 
 setup(
     name="airbyte-cdk",
-    version="0.1.100",
+    version="0.1.101",
     description="A framework for writing Airbyte Connectors.",
     long_description=README,
     long_description_content_type="text/markdown",

airbyte-cdk/python/unit_tests/sources/declarative/test_create_partial.py

@@ -2,7 +2,8 @@
 # Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 #
 
-from airbyte_cdk.sources.declarative.create_partial import create
+import pytest
+from airbyte_cdk.sources.declarative.create_partial import _key_is_unset_or_identical, create
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 
 
@@ -33,6 +34,12 @@ def test_pass_parameter_to_create_function():
     assert object.another_param == "B"
 
 
+def test_parameter_not_overwritten_by_options():
+    object = create(AClass, parameter="A", another_param="B", **{"$options": {"parameter": "C"}})()
+    assert object.parameter == "A"
+    assert object.another_param == "B"
+
+
 def test_overwrite_param():
     object = create(AClass, parameter="A", another_param="B")(parameter="C")
     assert object.parameter == "C"
@@ -46,7 +53,7 @@ def test_string_interpolation():
     assert interpolated_string.string == s
 
 
-def test_string_interpolation_through_kwargs():
+def test_string_interpolation_through_options():
     s = "{{ options['name'] }}"
     options = {"name": "airbyte"}
     partial = create(InterpolatedString, string=s, **options)
@@ -60,3 +67,17 @@ def test_string_interpolation_through_options_keyword():
     partial = create(InterpolatedString, string=s, **options)
     interpolated_string = partial()
     assert interpolated_string.eval({}) == "airbyte"
+
+
+@pytest.mark.parametrize(
+    "test_name, key, value, expected_result",
+    [
+        ("test", "key", "value", True),
+        ("test", "key", "a_different_value", False),
+        ("test", "a_different_key", "value", True),
+    ],
+)
+def test_key_is_unset_or_identical(test_name, key, value, expected_result):
+    mapping = {"key": "value"}
+    result = _key_is_unset_or_identical(key, value, mapping)
+    assert expected_result == result

airbyte-cdk/python/unit_tests/sources/declarative/test_factory.py

@@ -414,31 +414,31 @@ def test_create_record_selector(test_name, record_selector, expected_runtime_sel
         (
             "test_option_in_selector",
             """
-      extractor:
-        type: DpathExtractor
-        field_pointer: ["{{ options['name'] }}"]
-      selector:
-        class_name: airbyte_cdk.sources.declarative.extractors.record_selector.RecordSelector
-        $options:
-          name: "selector"
-        extractor: "*ref(extractor)"
-    """,
+          extractor:
+            type: DpathExtractor
+            field_pointer: ["{{ options['name'] }}"]
+          selector:
+            class_name: airbyte_cdk.sources.declarative.extractors.record_selector.RecordSelector
+            $options:
+              name: "selector"
+            extractor: "*ref(extractor)"
+        """,
             "selector",
         ),
         (
             "test_option_in_extractor",
             """
-      extractor:
-        type: DpathExtractor
-        $options:
-          name: "extractor"
-        field_pointer: ["{{ options['name'] }}"]
-      selector:
-        class_name: airbyte_cdk.sources.declarative.extractors.record_selector.RecordSelector
-        $options:
-          name: "selector"
-        extractor: "*ref(extractor)"
-    """,
+          extractor:
+            type: DpathExtractor
+            $options:
+              name: "extractor"
+            field_pointer: ["{{ options['name'] }}"]
+          selector:
+            class_name: airbyte_cdk.sources.declarative.extractors.record_selector.RecordSelector
+            $options:
+              name: "selector"
+            extractor: "*ref(extractor)"
+        """,
             "extractor",
         ),
     ],
@@ -682,6 +682,37 @@ def test_add_fields(self):
         ]
         assert expected == component.transformations
 
+    def test_add_fields_path_in_options(self):
+        content = f"""
+        the_stream:
+            class_name: airbyte_cdk.sources.declarative.declarative_stream.DeclarativeStream
+            $options:
+                {self.base_options}
+                path: "/wrong_path"
+                transformations:
+                    - type: AddFields
+                      fields:
+                        - path: ["field1"]
+                          value: "static_value"
+        """
+        config = parser.parse(content)
+
+        factory.create_component(config["the_stream"], input_config, False)
+
+        component = factory.create_component(config["the_stream"], input_config)()
+        assert isinstance(component, DeclarativeStream)
+        expected = [
+            AddFields(
+                fields=[
+                    AddedFieldDefinition(
+                        path=["field1"], value=InterpolatedString(string="static_value", default="static_value", options={}), options={}
+                    )
+                ],
+                options={},
+            )
+        ]
+        assert expected == component.transformations
+
 
 def test_validation_wrong_input_type():
     content = """