fix: strip null optional config fields before validation and include field path in error messages

Config validation now strips null-valued non-required fields before
running jsonschema validation. This prevents spurious 'None is not of
type string' errors when the platform passes null for unset optional
fields.

Additionally, validation error messages now include the JSON path to
the failing field, making config errors actionable for users.

Co-Authored-By: gl_anatolii.yatsuk <gl_anatolii.yatsuk@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/utils/schema_helpers.py

@@ -185,22 +185,38 @@ def _resolve_schema_references(self, raw_schema: dict[str, Any]) -> dict[str, An
             raise ValueError(f"Expected resolved to be a dict. Got {resolved}")
 
 
+def _strip_nulls_from_optional_fields(config: Mapping[str, Any], spec_schema: Mapping[str, Any]) -> dict[str, Any]:
+    """Remove keys with None values that are not required by the schema.
+
+    The platform may store optional config fields as explicit JSON null rather
+    than omitting the key. JSON Schema treats *absent* keys as valid for
+    non-required properties but rejects null unless the type includes "null".
+    Stripping these avoids spurious validation failures.
+    """
+    required_fields = set(spec_schema.get("required", []))
+    return {k: v for k, v in config.items() if v is not None or k in required_fields}
+
+
 def check_config_against_spec_or_exit(
     config: Mapping[str, Any], spec: ConnectorSpecification
 ) -> None:
-    """
-    Check config object against spec. In case of spec is invalid, throws
-    an exception with validation error description.
+    """Check config object against spec and raise on validation failure.
 
-    :param config - config loaded from file specified over command line
-    :param spec - spec object generated by connector
+    Strips null-valued optional fields before validation so that connectors
+    are not broken by platform-level null injection for unset fields.
     """
     spec_schema = spec.connectionSpecification
+    sanitized_config = _strip_nulls_from_optional_fields(config, spec_schema)
     try:
-        validate(instance=config, schema=spec_schema)
+        validate(instance=sanitized_config, schema=spec_schema)
     except ValidationError as validation_error:
+        field_path = ".".join(str(p) for p in validation_error.absolute_path)
+        if field_path:
+            message = f"Config validation error: '{field_path}' - {validation_error.message}"
+        else:
+            message = f"Config validation error: {validation_error.message}"
         raise AirbyteTracedException(
-            message="Config validation error: " + validation_error.message,
+            message=message,
             internal_message=validation_error.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

@@ -197,6 +197,57 @@ def test_shared_schemas_resolves_nested():
         assert jsonref.JsonRef.replace_refs(actual_schema)
 
 
+@pytest.mark.parametrize(
+    "config,spec_schema",
+    [
+        pytest.param(
+            {"api_token": "valid", "optional_field": None},
+            {"type": "object", "required": ["api_token"], "properties": {"api_token": {"type": "string"}, "optional_field": {"type": "string"}}},
+            id="null_optional_field_is_stripped",
+        ),
+        pytest.param(
+            {"api_token": "valid"},
+            {"type": "object", "required": ["api_token"], "properties": {"api_token": {"type": "string"}, "optional_field": {"type": "string"}}},
+            id="absent_optional_field_passes",
+        ),
+        pytest.param(
+            {"api_token": "valid", "start_date": None, "end_date": None, "app_id": None},
+            {"type": "object", "required": ["api_token"], "properties": {"api_token": {"type": "string"}, "start_date": {"type": "string"}, "end_date": {"type": "string"}, "app_id": {"type": "string"}}},
+            id="multiple_null_optional_fields_stripped",
+        ),
+    ],
+)
+def test_check_config_strips_null_optional_fields(config, spec_schema):
+    spec = ConnectorSpecificationSerializer.load({"connectionSpecification": spec_schema})
+    check_config_against_spec_or_exit(config, spec)
+
+
+@pytest.mark.parametrize(
+    "config,spec_schema,expected_substring",
+    [
+        pytest.param(
+            {"api_token": None, "optional_field": "present"},
+            {"type": "object", "required": ["api_token"], "properties": {"api_token": {"type": "string"}, "optional_field": {"type": "string"}}},
+            "api_token",
+            id="null_required_field_reports_field_name",
+        ),
+        pytest.param(
+            {"credentials": {"token": 123}},
+            {"type": "object", "required": ["credentials"], "properties": {"credentials": {"type": "object", "required": ["token"], "properties": {"token": {"type": "string"}}}}},
+            "credentials.token",
+            id="nested_error_includes_full_path",
+        ),
+    ],
+)
+def test_check_config_error_message_includes_field_path(config, spec_schema, expected_substring):
+    spec = ConnectorSpecificationSerializer.load({"connectionSpecification": spec_schema})
+    with pytest_raises(AirbyteTracedException) as exc_info:
+        check_config_against_spec_or_exit(config, spec)
+    assert exc_info.value.failure_type == FailureType.config_error
+    assert expected_substring in exc_info.value.message
+    assert "Config validation error:" in exc_info.value.message
+
+
 @pytest.mark.parametrize(
     "limit, record_count, expected",
     [