Support custom fallbacks

singer_sdk/connectors/sql.py

@@ -238,6 +238,8 @@ def __init__(self) -> None:
             "ipv6": lambda _: sa.types.VARCHAR(45),
         }
 
+        self._fallback_type = sa.types.VARCHAR
+
     def _invoke_handler(  # noqa: PLR6301
         self,
         handler: JSONtoSQLHandler,
@@ -256,6 +258,24 @@ def _invoke_handler(  # noqa: PLR6301
             return handler()  # type: ignore[no-any-return]
         return handler(schema)
 
+    @property
+    def fallback_type(self) -> type[sa.types.TypeEngine]:
+        """Return the fallback type.
+
+        Returns:
+            The fallback type.
+        """
+        return self._fallback_type
+
+    @fallback_type.setter
+    def fallback_type(self, value: type[sa.types.TypeEngine]) -> None:
+        """Set the fallback type.
+
+        Args:
+            value: The new fallback type.
+        """
+        self._fallback_type = value
+
     def register_type_handler(self, json_type: str, handler: JSONtoSQLHandler) -> None:
         """Register a custom type handler.
 
@@ -385,7 +405,7 @@ def to_sql_type(self, schema: dict) -> sa.types.TypeEngine:
                     return sql_type
 
         # Fallback
-        return sa.types.VARCHAR()
+        return self.fallback_type()
 
 
 class SQLConnector:  # noqa: PLR0904

tests/core/test_connector_sql.py

@@ -597,6 +597,13 @@ def test_anyof_unknown(self, json_schema_to_sql: JSONSchemaToSQL):
         result = json_schema_to_sql.to_sql_type(jsonschema_type)
         assert isinstance(result, sa.types.VARCHAR)
 
+    def test_custom_fallback(self):
+        json_schema_to_sql = JSONSchemaToSQL()
+        json_schema_to_sql.fallback_type = sa.types.CHAR
+        jsonschema_type = {"cannot": "compute"}
+        result = json_schema_to_sql.to_sql_type(jsonschema_type)
+        assert isinstance(result, sa.types.CHAR)
+
     @pytest.mark.parametrize(
         "jsonschema_type,expected_type",
         [