feat(taps): In OpenAPI schema normalization, treat all fields as nullable unless they are part of a primary key (#3369)

## Summary by Sourcery

Treat all fields as nullable by default in OpenAPI schema normalization,
marking only primary key properties as non-nullable by passing
key_properties through preprocessing and normalization flows

New Features:
- Add key_properties parameter to schema preprocessing and normalization
methods to distinguish primary keys

Enhancements:
- Propagate key_properties (stream.primary_keys) through SchemaSource
and StreamSchema descriptor to schema preprocessing

Tests:
- Update tests to expect default nullable types for all fields except
primary keys

Author: edgarrmondragon

singer_sdk/schema/source.py

@@ -34,7 +34,7 @@
     from typing_extensions import TypeVar
 
 if t.TYPE_CHECKING:
-    from collections.abc import Callable
+    from collections.abc import Callable, Sequence
 
     from singer_sdk.helpers._compat import Traversable
     from singer_sdk.streams.core import Stream
@@ -64,11 +64,17 @@ class SchemaPreprocessor(t.Protocol):
     Implementations can normalize, validate, or modify schemas as needed.
     """
 
-    def preprocess_schema(self, schema: Schema) -> Schema:
+    def preprocess_schema(
+        self,
+        schema: Schema,
+        *,
+        key_properties: Sequence[str] = (),
+    ) -> Schema:
         """Pre-process a schema.
 
         Args:
             schema: A JSON schema to preprocess.
+            key_properties: The stream's key properties.
 
         Returns:
             The preprocessed schema.
@@ -84,25 +90,41 @@ def __init__(self) -> None:
         self._preprocessor: SchemaPreprocessor | None = None
 
     @t.final
-    def preprocess_schema(self, schema: Schema) -> Schema:
+    def preprocess_schema(
+        self,
+        schema: Schema,
+        *,
+        key_properties: Sequence[str] = (),
+    ) -> Schema:
         """Pre-process the schema before providing it to the stream.
 
         Args:
             schema: A JSON schema.
+            key_properties: The stream's key properties.
 
         Returns:
             The pre-processed schema.
         """
         if self._preprocessor is None:
             return schema
-        return self._preprocessor.preprocess_schema(schema)
+        return self._preprocessor.preprocess_schema(
+            schema,
+            key_properties=key_properties,
+        )
 
     @t.final
-    def get_schema(self, key: _TKey, /) -> Schema:
+    def get_schema(
+        self,
+        key: _TKey,
+        /,
+        *,
+        key_properties: Sequence[str] = (),
+    ) -> Schema:
         """Convenience method to get a schema component.
 
         Args:
             key: The schema component name to retrieve.
+            key_properties: The stream's key properties.
 
         Returns:
             A JSON schema dictionary.
@@ -122,7 +144,10 @@ def get_schema(self, key: _TKey, /) -> Schema:
                 raise SchemaNotValidError(msg)
             self._schema_cache[key] = schema
 
-        return self.preprocess_schema(self._schema_cache[key])
+        return self.preprocess_schema(
+            self._schema_cache[key],
+            key_properties=key_properties,
+        )
 
     @abstractmethod
     def fetch_schema(self, key: _TKey) -> Schema:
@@ -203,7 +228,10 @@ def get_stream_schema(self, stream: Stream, stream_class: type[Stream]) -> Schem
         Returns:
             A JSON schema dictionary.
         """
-        return self.schema_source.get_schema(self.key or stream.name)  # type: ignore[arg-type]
+        return self.schema_source.get_schema(
+            self.key or stream.name,  # type: ignore[arg-type]
+            key_properties=stream.primary_keys,
+        )
 
 
 def _load_yaml(content: bytes) -> dict[str, t.Any]:
@@ -223,16 +251,24 @@ class OpenAPISchemaNormalizer(SchemaPreprocessor):
     - Recursively processes nested object properties and array items
     """
 
-    def handle_object(self, schema: Schema) -> Schema:
+    def handle_object(
+        self,
+        schema: Schema,
+        *,
+        key_properties: Sequence[str] = (),
+    ) -> Schema:
         """Handle JSON object schemas.
 
         Args:
             schema: A JSON schema.
+            key_properties: The stream's key properties.
 
         Returns:
             The processed object schema.
         """
+        schema["nullable"] = len(key_properties) == 0
         for prop, prop_schema in schema.get("properties", {}).items():
+            prop_schema["nullable"] = prop not in key_properties
             schema["properties"][prop] = self.normalize_schema(prop_schema)
         return schema
 
@@ -290,14 +326,20 @@ def handle_all_of(self, subschemas: list[Schema]) -> Schema:  # noqa: PLR6301
 
         return result
 
-    def normalize_schema(self, schema: Schema) -> Schema:
+    def normalize_schema(
+        self,
+        schema: Schema,
+        *,
+        key_properties: Sequence[str] = (),
+    ) -> Schema:
         """Normalize an OpenAPI schema to standard JSON Schema.
 
         This method applies a series of transformations to convert OpenAPI-specific
         schema features (like 'nullable') into standard JSON Schema constructs.
 
         Args:
             schema: The schema to normalize.
+            key_properties: The stream's key properties.
 
         Returns:
             A normalized schema dictionary.
@@ -306,7 +348,7 @@ def normalize_schema(self, schema: Schema) -> Schema:
         schema_type: str | list[str] = result.get("type", [])
 
         if "object" in schema_type:
-            result = self.handle_object(result)
+            result = self.handle_object(result, key_properties=key_properties)
 
         elif "array" in schema_type and (items := result.get("items")):
             result["items"] = self.handle_array_items(items)
@@ -330,8 +372,13 @@ def normalize_schema(self, schema: Schema) -> Schema:
         return result
 
     @override
-    def preprocess_schema(self, schema: Schema) -> Schema:
-        return self.normalize_schema(schema)
+    def preprocess_schema(
+        self,
+        schema: Schema,
+        *,
+        key_properties: Sequence[str] = (),
+    ) -> Schema:
+        return self.normalize_schema(schema, key_properties=key_properties)
 
 
 class OpenAPISchema(SchemaSource[_TKey]):

tests/core/schema/test_source.py

@@ -45,8 +45,8 @@ def resolved_user_schema() -> dict[str, t.Any]:
         "type": "object",
         "properties": {
             "id": {"type": "string"},
-            "name": {"type": "string"},
-            "email": {"type": "string", "format": "email"},
+            "name": {"type": ["string", "null"]},
+            "email": {"type": ["string", "null"], "format": "email"},
         },
         "required": ["id", "name"],
     }
@@ -410,7 +410,7 @@ def test_openapi_component(
         openapi_file.write_text(content)
 
         source = OpenAPISchema(openapi_file)
-        result = source.get_schema("User")
+        result = source.get_schema("User", key_properties=("id",))
         assert result == resolved_user_schema
 
     def test_openapi_unknown_spec(
@@ -475,7 +475,7 @@ def test_openapi_schema_with_traversable(
         path.write_text(json.dumps(openapi_spec))
 
         source = OpenAPISchema(path)
-        result = source.get_schema("User")
+        result = source.get_schema("User", key_properties=("id",))
         assert result == resolved_user_schema
 
     def test_openapi_schema_with_unknown_file_type(
@@ -568,11 +568,11 @@ def get_unresolved_schema(self, key: PathAndMethodKey) -> dict[str, t.Any]:
         expected_schema = {
             "type": "array",
             "items": {
-                "type": "object",
+                "type": ["object", "null"],
                 "properties": {
-                    "id": {"type": "string"},
-                    "name": {"type": "string"},
-                    "email": {"type": "string", "format": "email"},
+                    "id": {"type": ["string", "null"]},
+                    "name": {"type": ["string", "null"]},
+                    "email": {"type": ["string", "null"], "format": "email"},
                 },
                 "required": ["id", "name"],
             },
@@ -630,6 +630,7 @@ def test_stream_schema_descriptor(
         class FooStream:
             name = "foo"
             schema: t.ClassVar[StreamSchema] = StreamSchema(schema_source)
+            primary_keys = ("id",)
 
         stream = FooStream()
         assert stream.schema == foo_schema
@@ -644,6 +645,7 @@ def test_stream_schema_descriptor_with_explicit_key(
         class BarStream:
             name = "bar"
             schema: t.ClassVar[StreamSchema] = StreamSchema(schema_source, key="foo")
+            primary_keys = ("id",)
 
         stream = BarStream()
         assert stream.schema == foo_schema
@@ -657,6 +659,7 @@ def test_stream_schema_descriptor_key_not_found(
         class BarStream:
             name = "bar"
             schema: t.ClassVar[StreamSchema] = StreamSchema(schema_source)
+            primary_keys = ("id",)
 
         stream = BarStream()
         with pytest.raises(
@@ -804,14 +807,14 @@ def test_normalize(
             schema = source.get_schema("AllOfSchemas")
             normalized = source.preprocess_schema(schema)
             assert normalized == {
-                "type": "object",
+                "type": ["object", "null"],
                 "properties": {
                     "references": {
-                        "type": "object",
+                        "type": ["object", "null"],
                         "additionalProperties": {"type": "string"},
                     },
-                    "created": {"type": "string"},
-                    "name": {"type": "string"},
+                    "created": {"type": ["string", "null"]},
+                    "name": {"type": ["string", "null"]},
                 },
             }
             assert "allOf" not in normalized