refactor: Added a class method to instantiate JSONSchemaToSQL from …

…the tap configuration

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

@@ -276,6 +276,35 @@ def __init__(self, *, max_varchar_length: int | None = None) -> None:
 
         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,
@@ -487,6 +516,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 +571,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]:

tests/core/test_connector_sql.py

@@ -654,7 +654,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 +668,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)