fix: include field path in config validation error messages

airbyte_cdk/sources/utils/schema_helpers.py

@@ -199,9 +199,26 @@ def check_config_against_spec_or_exit(
     try:
         validate(instance=config, schema=spec_schema)
     except ValidationError as validation_error:
+        field_path = (
+            ".".join(str(p) for p in validation_error.absolute_path)
+            if validation_error.absolute_path
+            else ""
+        )
+        if field_path:
+            user_message = (
+                f'Config validation error: Field "{field_path}" - {validation_error.message}.'
+            )
+        else:
+            user_message = f"Config validation error: {validation_error.message}."
+        schema_path = (
+            ".".join(str(p) for p in validation_error.absolute_schema_path)
+            if validation_error.absolute_schema_path
+            else ""
+        )
+        internal_message = f"{validation_error.message} (path: {field_path or '<root>'}, schema_path: {schema_path})"
         raise AirbyteTracedException(
-            message="Config validation error: " + validation_error.message,
-            internal_message=validation_error.message,
+            message=user_message,
+            internal_message=internal_message,
             failure_type=FailureType.config_error,
         ) from None  # required to prevent logging config secrets from the ValidationError's stacktrace
 

unit_tests/sources/utils/test_schema_helpers.py

@@ -81,6 +81,62 @@ def test_should_not_fail_validation_for_valid_config(spec_object):
     assert True, "should pass validation with valid config"
 
 
+@pytest.mark.parametrize(
+    "schema,config,expected_message_contains,expected_field_path",
+    [
+        pytest.param(
+            {
+                "type": "object",
+                "properties": {"api_key": {"type": "string"}},
+            },
+            {"api_key": None},
+            'Field "api_key"',
+            "api_key",
+            id="top_level_field_wrong_type",
+        ),
+        pytest.param(
+            {
+                "type": "object",
+                "properties": {
+                    "auth": {
+                        "type": "object",
+                        "properties": {"token": {"type": "string"}},
+                    }
+                },
+            },
+            {"auth": {"token": None}},
+            'Field "auth.token"',
+            "auth.token",
+            id="nested_field_wrong_type",
+        ),
+        pytest.param(
+            {
+                "type": "object",
+                "required": ["api_key"],
+                "properties": {"api_key": {"type": "string"}},
+            },
+            {},
+            "Config validation error:",
+            "",
+            id="missing_required_field_no_path",
+        ),
+    ],
+)
+def test_check_config_error_message_includes_field_path(
+    schema, config, expected_message_contains, expected_field_path
+):
+    spec = ConnectorSpecificationSerializer.load({"connectionSpecification": schema})
+
+    with pytest_raises(AirbyteTracedException) as ex_info:
+        check_config_against_spec_or_exit(config, spec)
+
+    exc = ex_info.value
+    assert expected_message_contains in exc.message
+    assert exc.failure_type == FailureType.config_error
+    if expected_field_path:
+        assert expected_field_path in exc.internal_message
+
+
 class TestResourceSchemaLoader:
     # Test that a simple schema is loaded correctly
     @staticmethod