fix(low-code): handle ScannerError in ConfigComponentsResolver (#614)

Co-authored-by: octavia-squidington-iii <[email protected]>

Author: darynaishchenko

airbyte_cdk/sources/declarative/resolvers/config_components_resolver.py

@@ -11,6 +11,7 @@
 import yaml
 from typing_extensions import deprecated
 from yaml.parser import ParserError
+from yaml.scanner import ScannerError
 
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.resolvers.components_resolver import (
@@ -185,4 +186,8 @@ def _parse_yaml_if_possible(value: Any) -> Any:
                 return yaml.safe_load(value)
             except ParserError:  # "{{ record[0] in ['cohortActiveUsers'] }}"   # not valid YAML
                 return value
+            except ScannerError as e:  # "%Y-%m-%d'   # not valid yaml
+                if "expected alphabetic or numeric character, but found '%'" in str(e):
+                    return value
+                raise e
         return value

unit_tests/sources/declarative/resolvers/test_config_components_resolver.py

@@ -145,6 +145,20 @@ def to_configured_catalog(
     STREAM_CONFIG
 ]
 
+# Manifest with component definition with value that is fails when trying
+# to parse yaml in _parse_yaml_if_possible but generally contains valid string
+_MANIFEST_WITH_SCANNER_ERROR = deepcopy(_MANIFEST)
+_MANIFEST_WITH_SCANNER_ERROR["dynamic_streams"][0]["components_resolver"][
+    "components_mapping"
+].append(
+    {
+        "type": "ComponentMappingDefinition",
+        "create_or_update": True,
+        "field_path": ["retriever", "requester", "$parameters", "cursor_format"],
+        "value": "{{ '%Y-%m-%d' if components_values['name'] == 'default_item' else '%Y-%m-%dT%H:%M:%S' }}",
+    }
+)
+
 
 @pytest.mark.parametrize(
     "manifest, config, expected_exception, expected_stream_names",
@@ -157,6 +171,7 @@ def to_configured_catalog(
             None,
         ),
         (_MANIFEST_WITH_STREAM_CONFIGS_LIST, _CONFIG, None, ["item_1", "item_2", "default_item"]),
+        (_MANIFEST_WITH_SCANNER_ERROR, _CONFIG, None, ["item_1", "item_2", "default_item"]),
     ],
 )
 def test_dynamic_streams_read_with_config_components_resolver(