feat(taps,targets): Support the x-singer.decimal JSON Schema extension

docs/guides/sql-target.md

@@ -30,7 +30,10 @@ def custom_array_to_sql(jsonschema: dict) -> VectorType | sa.types.VARCHAR:
 class MyConnector(SQLConnector):
     @functools.cached_property
     def jsonschema_to_sql(self):
-        to_sql = JSONSchemaToSQL()
+        to_sql = JSONSchemaToSQL.from_config(
+            self.config,
+            max_varchar_length=self.max_varchar_length,
+        )
         to_sql.register_type_handler("array", custom_array_to_sql)
         return to_sql
 ```
@@ -46,7 +49,10 @@ from my_sqlalchemy_dialect import URI
 class MyConnector(SQLConnector):
     @functools.cached_property
     def jsonschema_to_sql(self):
-        to_sql = JSONSchemaToSQL()
+        to_sql = JSONSchemaToSQL.from_config(
+            self.config,
+            max_varchar_length=self.max_varchar_length,
+        )
         to_sql.register_format_handler("uri", URI)
         return to_sql
 ```

singer_sdk/connectors/sql.py

@@ -124,10 +124,21 @@ class SQLToJSONSchema:
     .. versionchanged:: 0.43.0
        Added the :meth:`singer_sdk.connectors.sql.SQLToJSONSchema.from_config` class
        method.
+    .. versionchanged:: 0.43.0
+       Added support for the `use_singer_decimal` option.
     """
 
+    def __init__(self, *, use_singer_decimal: bool = False) -> None:
+        """Initialize the SQL to JSON Schema converter.
+
+        Args:
+            use_singer_decimal: Whether to represent numbers as `string` with
+                the `singer.decimal` format instead of as `number`.
+        """
+        self.use_singer_decimal = use_singer_decimal
+
     @classmethod
-    def from_config(cls: type[SQLToJSONSchema], config: dict) -> SQLToJSONSchema:  # noqa: ARG003
+    def from_config(cls: type[SQLToJSONSchema], config: dict) -> SQLToJSONSchema:
         """Create a new instance from a configuration dictionary.
 
         Override this to instantiate this converter with values from the tap's
@@ -146,11 +157,13 @@ def from_config(cls, config):
 
         Args:
             config: The configuration dictionary.
+            use_singer_decimal: Whether to represent numbers as `string` with
+                the `singer.decimal` format instead of as `number`.
 
         Returns:
             A new instance of the class.
         """
-        return cls()
+        return cls(use_singer_decimal=config.get("use_singer_decimal", False))
 
     @functools.singledispatchmethod
     def to_jsonschema(self, column_type: sa.types.TypeEngine) -> dict:  # noqa: ARG002, D102, PLR6301
@@ -193,12 +206,14 @@ def integer_to_jsonschema(self, column_type: sa.types.Integer) -> dict:  # noqa:
         return th.IntegerType.type_dict  # type: ignore[no-any-return]
 
     @to_jsonschema.register
-    def float_to_jsonschema(self, column_type: sa.types.Numeric) -> dict:  # noqa: ARG002, PLR6301
+    def float_to_jsonschema(self, column_type: sa.types.Numeric) -> dict:  # noqa: ARG002
         """Return a JSON Schema representation of a generic number type.
 
         Args:
             column_type (:column_type:`Numeric`): The column type.
         """
+        if self.use_singer_decimal:
+            return th.SingerDecimalType.type_dict  # type: ignore[no-any-return]
         return th.NumberType.type_dict  # type: ignore[no-any-return]
 
     @to_jsonschema.register
@@ -272,10 +287,40 @@ def __init__(self, *, max_varchar_length: int | None = None) -> None:
             "hostname": lambda _: sa.types.VARCHAR(253),  # RFC 1035
             "ipv4": lambda _: sa.types.VARCHAR(15),
             "ipv6": lambda _: sa.types.VARCHAR(45),
+            "singer.decimal": self._handle_singer_decimal,
         }
 
         self._fallback_type: type[sa.types.TypeEngine] = sa.types.VARCHAR
 
+    @classmethod
+    def from_config(
+        cls: type[JSONSchemaToSQL],
+        config: dict,  # noqa: ARG003
+        *,
+        max_varchar_length: int | None,
+    ) -> JSONSchemaToSQL:
+        """Create a new instance from a configuration dictionary.
+
+        Override this to instantiate this converter with values from the target's
+        configuration dictionary.
+
+        .. code-block:: python
+
+              class CustomJSONSchemaToSQL(JSONSchemaToSQL):
+                  @classmethod
+                  def from_config(cls, config, **kwargs):
+                      return cls(max_varchar_length=config.get("max_varchar_length"))
+
+        Args:
+            config: The configuration dictionary.
+            max_varchar_length: The absolute maximum length for VARCHAR columns that
+                the database supports.
+
+        Returns:
+            A new instance of the class.
+        """
+        return cls(max_varchar_length=max_varchar_length)
+
     def _invoke_handler(  # noqa: PLR6301
         self,
         handler: JSONtoSQLHandler,
@@ -294,6 +339,17 @@ def _invoke_handler(  # noqa: PLR6301
             return handler()  # type: ignore[no-any-return]
         return handler(schema)
 
+    def _handle_singer_decimal(self, schema: dict) -> sa.types.TypeEngine:  # noqa: PLR6301
+        """Handle a singer.decimal format.
+
+        Args:
+            schema: The JSON Schema object.
+
+        Returns:
+            The appropriate SQLAlchemy type.
+        """
+        return sa.types.DECIMAL(schema.get("precision"), schema.get("scale"))
+
     @property
     def fallback_type(self) -> type[sa.types.TypeEngine]:
         """Return the fallback type.
@@ -487,6 +543,11 @@ class SQLConnector:  # noqa: PLR0904
     #: a custom mapping for your SQL dialect.
     sql_to_jsonschema_converter: type[SQLToJSONSchema] = SQLToJSONSchema
 
+    #: The JSON-to-SQL type mapper class for this SQL connector. Override this property
+    #: with a subclass of :class:`~singer_sdk.connectors.sql.JSONSchemaToSQL` to provide
+    #: a custom mapping for your SQL dialect.
+    jsonschema_to_sql_converter: type[JSONSchemaToSQL] = JSONSchemaToSQL
+
     def __init__(
         self,
         config: dict | None = None,
@@ -537,7 +598,10 @@ def jsonschema_to_sql(self) -> JSONSchemaToSQL:
 
         .. versionadded:: 0.42.0
         """
-        return JSONSchemaToSQL(max_varchar_length=self.max_varchar_length)
+        return self.jsonschema_to_sql_converter.from_config(
+            self.config,
+            max_varchar_length=self.max_varchar_length,
+        )
 
     @contextmanager
     def _connect(self) -> t.Iterator[sa.engine.Connection]:

singer_sdk/typing.py

@@ -448,6 +448,12 @@ class RegexType(StringType):
     string_format = "regex"
 
 
+class SingerDecimalType(StringType):
+    """Decimal type."""
+
+    string_format = "singer.decimal"
+
+
 class BooleanType(JSONTypeHelper[bool]):
     """Boolean type.
 

tests/core/test_connector_sql.py

@@ -482,6 +482,14 @@ def my_type_to_jsonschema(self, column_type) -> dict:  # noqa: ARG002
     assert m.to_jsonschema(sa.types.BOOLEAN()) == {"type": ["boolean"]}
 
 
+def test_numeric_to_singer_decimal():
+    converter = SQLToJSONSchema(use_singer_decimal=True)
+    assert converter.to_jsonschema(sa.types.NUMERIC()) == {
+        "type": ["string"],
+        "format": "singer.decimal",
+    }
+
+
 class TestJSONSchemaToSQL:  # noqa: PLR0904
     @pytest.fixture
     def json_schema_to_sql(self) -> JSONSchemaToSQL:
@@ -654,7 +662,7 @@ def test_unknown_format(self, json_schema_to_sql: JSONSchemaToSQL):
         assert isinstance(result, sa.types.VARCHAR)
 
     def test_custom_fallback(self):
-        json_schema_to_sql = JSONSchemaToSQL()
+        json_schema_to_sql = JSONSchemaToSQL(max_varchar_length=None)
         json_schema_to_sql.fallback_type = sa.types.CHAR
         jsonschema_type = {"cannot": "compute"}
         result = json_schema_to_sql.to_sql_type(jsonschema_type)
@@ -668,7 +676,7 @@ def handle_raw_string(self, schema):
 
                 return super().handle_raw_string(schema)
 
-        json_schema_to_sql = CustomJSONSchemaToSQL()
+        json_schema_to_sql = CustomJSONSchemaToSQL(max_varchar_length=None)
 
         vanilla = {"type": ["string"]}
         result = json_schema_to_sql.to_sql_type(vanilla)
@@ -688,3 +696,9 @@ def handle_raw_string(self, schema):
         }
         result = json_schema_to_sql.to_sql_type(image_type)
         assert isinstance(result, sa.types.LargeBinary)
+
+    def test_singer_decimal(self):
+        json_schema_to_sql = JSONSchemaToSQL()
+        jsonschema_type = {"type": ["string"], "format": "singer.decimal"}
+        result = json_schema_to_sql.to_sql_type(jsonschema_type)
+        assert isinstance(result, sa.types.DECIMAL)