feat(dynamic-streams): Avoid parsing strings in ConfigComponentResolver

airbyte_cdk/sources/declarative/resolvers/config_components_resolver.py

@@ -177,7 +177,11 @@ def resolve_components(
                 )
 
                 path = [path.eval(self.config, **kwargs) for path in resolved_component.field_path]
-                parsed_value = self._parse_yaml_if_possible(value)
+                # Avoid parsing strings that are meant to be strings
+                if not (isinstance(value, str) and valid_types == (str,)):
+                    parsed_value = self._parse_yaml_if_possible(value)
+                else:
+                    parsed_value = value
                 updated = dpath.set(updated_config, path, parsed_value)
 
                 if parsed_value and not updated and resolved_component.create_or_update:

unit_tests/sources/declarative/resolvers/test_config_components_resolver.py

@@ -4,6 +4,7 @@
 
 import json
 from copy import deepcopy
+from datetime import date as dt_date
 from unittest.mock import MagicMock
 
 import pytest
@@ -386,3 +387,93 @@ def test_component_mapping_conditions(manifest, config, expected_conditional_par
                 stream._stream_partition_generator._partition_factory._retriever.requester._parameters
                 == expected_conditional_params[stream.name]
             )
+
+
+_MANIFEST_WITH_VALUE_TYPE_STR = deepcopy(_MANIFEST)
+_MANIFEST_WITH_VALUE_TYPE_STR["dynamic_streams"][0]["components_resolver"][
+    "components_mapping"
+].extend(
+    [
+        {
+            "type": "ComponentMappingDefinition",
+            "field_path": ["retriever", "requester", "$parameters", "as_string"],
+            "value": "true",  # valid YAML, but we want to keep it as *string*
+            "value_type": "string",
+            "create_or_update": True,
+        },
+        {
+            "type": "ComponentMappingDefinition",
+            "field_path": ["retriever", "requester", "$parameters", "as_yaml"],
+            "value": "true",  # no value_type -> should be parsed to boolean True
+            "create_or_update": True,
+        },
+        {
+            "type": "ComponentMappingDefinition",
+            "field_path": ["retriever", "requester", "$parameters", "json_string"],
+            "value": "[1, 2]",  # valid YAML/JSON-looking text; keep as string
+            "value_type": "string",
+            "create_or_update": True,
+        },
+        {
+            "type": "ComponentMappingDefinition",
+            "field_path": ["retriever", "requester", "$parameters", "json_parsed"],
+            "value": "[1, 2]",  # no value_type -> should parse to a list
+            "create_or_update": True,
+        },
+        {
+            "type": "ComponentMappingDefinition",
+            "field_path": ["retriever", "requester", "$parameters", "date_as_string"],
+            "value": "2024-07-10",  # date-like text that YAML would parse; force keep as string
+            "value_type": "string",
+            "create_or_update": True,
+        },
+        {
+            "type": "ComponentMappingDefinition",
+            "field_path": ["retriever", "requester", "$parameters", "date_yaml_parsed"],
+            "value": "2024-07-10",  # no value_type -> YAML should parse to datetime.date
+            "create_or_update": True,
+        },
+    ]
+)
+
+
+def test_value_type_str_avoids_yaml_parsing():
+    source = ConcurrentDeclarativeSource(
+        source_config=_MANIFEST_WITH_VALUE_TYPE_STR, config=_CONFIG, catalog=None, state=None
+    )
+
+    for stream in source.streams(_CONFIG):
+        params = (
+            stream._stream_partition_generator._partition_factory._retriever.requester._parameters
+        )
+
+        # Confirm the usual param is still present
+        assert "item_id" in params
+
+        # value_type="string" -> keep as string
+        assert "as_string" in params
+        assert isinstance(params["as_string"], str)
+        assert params["as_string"] == "true"
+
+        assert "json_string" in params
+        assert isinstance(params["json_string"], str)
+        assert params["json_string"] == "[1, 2]"
+
+        # No value_type -> YAML parsed
+        assert "as_yaml" in params
+        assert isinstance(params["as_yaml"], bool)
+        assert params["as_yaml"] is True
+
+        assert "json_parsed" in params
+        assert isinstance(params["json_parsed"], list)
+        assert params["json_parsed"] == [1, 2]
+
+        # value_type="string" -> remains a plain string
+        assert "date_as_string" in params
+        assert isinstance(params["date_as_string"], str)
+        assert params["date_as_string"] == "2024-07-10"
+
+        # no value_type -> YAML parses to datetime.date
+        assert "date_yaml_parsed" in params
+        assert isinstance(params["date_yaml_parsed"], dt_date)
+        assert params["date_yaml_parsed"].isoformat() == "2024-07-10"