Test the output

singer_sdk/_singerlib/catalog.py

@@ -7,10 +7,6 @@
 
 from singer_sdk._singerlib.schema import Schema
 
-if t.TYPE_CHECKING:
-    from typing_extensions import TypeAlias
-
-
 Breadcrumb = tuple[str, ...]
 
 logger = logging.getLogger(__name__)
@@ -35,7 +31,7 @@ def __missing__(self, breadcrumb: Breadcrumb) -> bool:
 
 
 @dataclass
-class Metadata:
+class _BaseMetadata:
     """Base stream or property metadata."""
 
     class InclusionType(str, enum.Enum):
@@ -50,7 +46,7 @@ class InclusionType(str, enum.Enum):
     selected_by_default: bool | None = None
 
     @classmethod
-    def from_dict(cls: type[Metadata], value: dict[str, t.Any]) -> Metadata:
+    def from_dict(cls: type[_BaseMetadata], value: dict[str, t.Any]) -> _BaseMetadata:
         """Parse metadata dictionary.
 
         Args:
@@ -82,6 +78,11 @@ def to_dict(self) -> dict[str, t.Any]:
         return result
 
 
+@dataclass
+class Metadata(_BaseMetadata):
+    sql_datatype: str | None = None
+
+
 @dataclass
 class StreamMetadata(Metadata):
     """Stream metadata."""
@@ -93,15 +94,7 @@ class StreamMetadata(Metadata):
     schema_name: str | None = None
 
 
-@dataclass
-class SQLMetadata(Metadata):
-    sql_datatype: str | None = None
-
-
-AnyMetadata: TypeAlias = t.Union[Metadata, StreamMetadata]
-
-
-class MetadataMapping(dict[Breadcrumb, AnyMetadata]):
+class MetadataMapping(dict[Breadcrumb, _BaseMetadata]):
     """Stream metadata mapping."""
 
     @classmethod
@@ -138,7 +131,7 @@ def to_list(self) -> list[dict[str, t.Any]]:
             {"breadcrumb": list(k), "metadata": v.to_dict()} for k, v in self.items()
         ]
 
-    def __missing__(self, breadcrumb: Breadcrumb) -> AnyMetadata:
+    def __missing__(self, breadcrumb: Breadcrumb) -> _BaseMetadata:
         """Handle missing metadata entries.
 
         Args:
@@ -169,7 +162,7 @@ def get_standard_metadata(
         valid_replication_keys: list[str] | None = None,
         replication_method: str | None = None,
         selected_by_default: bool | None = None,
-        sql_datatypes: dict | None = None,
+        sql_datatypes: dict[str, str] | None = None,
     ) -> MetadataMapping:
         """Get default metadata for a stream.
 
@@ -200,17 +193,23 @@ def get_standard_metadata(
                 root.schema_name = schema_name
 
             for field_name in schema.get("properties", {}):
-                if sql_datatypes is None:
-                    entry = Metadata()
+                if sql_datatypes and field_name in sql_datatypes:
+                    sql_datatype = sql_datatypes[field_name]
                 else:
-                    entry = SQLMetadata(sql_datatype=sql_datatypes.get(field_name))
+                    sql_datatype = None
 
                 if (key_properties and field_name in key_properties) or (
                     valid_replication_keys and field_name in valid_replication_keys
                 ):
-                    entry.inclusion = Metadata.InclusionType.AUTOMATIC
+                    entry = Metadata(
+                        inclusion=Metadata.InclusionType.AUTOMATIC,
+                        sql_datatype=sql_datatype,
+                    )
                 else:
-                    entry.inclusion = Metadata.InclusionType.AVAILABLE
+                    entry = Metadata(
+                        inclusion=Metadata.InclusionType.AVAILABLE,
+                        sql_datatype=sql_datatype,
+                    )
 
                 mapping["properties", field_name] = entry
 

singer_sdk/connectors/sql.py

@@ -905,7 +905,7 @@ def discover_catalog_entry(
 
         # Initialize columns list
         table_schema = th.PropertiesList()
-        datatypes = {}
+        datatypes: dict[str, str] = {}
         for column_def in inspected.get_columns(table_name, schema=schema_name):
             column_name = column_def["name"]
             is_nullable = column_def.get("nullable", False)

tests/samples/test_tap_sqlite.py

@@ -11,7 +11,7 @@
 from samples.sample_tap_sqlite import SQLiteTap
 from samples.sample_target_csv.csv_target import SampleTargetCSV
 from singer_sdk import SQLStream
-from singer_sdk._singerlib import MetadataMapping, StreamMetadata
+from singer_sdk._singerlib import Metadata, MetadataMapping, StreamMetadata
 from singer_sdk.testing import (
     get_standard_tap_tests,
     tap_sync_test,
@@ -83,6 +83,11 @@ def test_sqlite_discovery(sqlite_sample_tap: SQLTap):
 
     assert stream.metadata.root.table_key_properties == ["c1"]
     assert stream.primary_keys == ["c1"]
+
+    field_metadata = stream.metadata["properties", "c1"]
+    assert isinstance(field_metadata, Metadata)
+    assert field_metadata.sql_datatype == "INTEGER()"
+
     assert stream.schema["properties"]["c1"] == {"type": ["integer"]}
     assert stream.schema["required"] == ["c1"]