feat: In SQL targets, users can now disable column type alterations with the allow_column_alter built-in setting

samples/sample_duckdb/connector.py

@@ -6,8 +6,6 @@
 
 
 class DuckDBConnector(SQLConnector):
-    allow_column_alter = True
-
     @staticmethod
     def get_column_alter_ddl(
         table_name: str,

singer_sdk/connectors/sql.py

@@ -15,7 +15,10 @@
 from singer_sdk._singerlib import CatalogEntry, MetadataMapping, Schema
 from singer_sdk.exceptions import ConfigValidationError
 from singer_sdk.helpers._util import dump_json, load_json
-from singer_sdk.helpers.capabilities import TargetLoadMethods
+from singer_sdk.helpers.capabilities import (
+    TARGET_ALLOW_COLUMN_ALTER_CONFIG,
+    TargetLoadMethods,
+)
 
 if t.TYPE_CHECKING:
     from sqlalchemy.engine import Engine
@@ -37,7 +40,10 @@ class SQLConnector:  # noqa: PLR0904
 
     allow_column_add: bool = True  # Whether ADD COLUMN is supported.
     allow_column_rename: bool = True  # Whether RENAME COLUMN is supported.
-    allow_column_alter: bool = False  # Whether altering column types is supported.
+
+    #: Whether altering column types is supported.
+    allow_column_alter = TARGET_ALLOW_COLUMN_ALTER_CONFIG.attribute(default=False)
+
     allow_merge_upsert: bool = False  # Whether MERGE UPSERT is supported.
     allow_overwrite: bool = False  # Whether overwrite load method is supported.
     allow_temp_tables: bool = True  # Whether temp tables are supported.
@@ -1125,7 +1131,7 @@ def _adapt_column_type(
             return
 
         # Not the same type, generic type or compatible types
-        # calling merge_sql_types for assistnace
+        # calling merge_sql_types for assistance
         compatible_sql_type = self.merge_sql_types([current_type, sql_type])
 
         if str(compatible_sql_type) == str(current_type):

singer_sdk/helpers/_config_property.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import typing as t
+
+T = t.TypeVar("T")
+
+
+class ConfigProperty(t.Generic[T]):
+    """A descriptor that gets a value from a named key of the config attribute."""
+
+    def __init__(self, custom_key: str | None = None, *, default: T | None = None):
+        """Initialize the descriptor.
+
+        Args:
+            custom_key: The key to get from the config attribute instead of the
+                attribute name.
+            default: The default value if the key is not found.
+        """
+        self.key = custom_key
+        self.default = default
+
+    def __set_name__(self, owner, name: str) -> None:  # noqa: ANN001
+        """Set the name of the attribute.
+
+        Args:
+            owner: The class of the object.
+            name: The name of the attribute.
+        """
+        self.key = self.key or name
+
+    def __get__(self, instance, owner) -> T | None:  # noqa: ANN001
+        """Get the value from the instance's config attribute.
+
+        Args:
+            instance: The instance of the object.
+            owner: The class of the object.
+
+        Returns:
+            The value from the config attribute.
+        """
+        return instance.config.get(self.key, self.default)  # type: ignore[no-any-return]

singer_sdk/helpers/capabilities.py

@@ -6,6 +6,7 @@
 from enum import Enum, EnumMeta
 from warnings import warn
 
+from singer_sdk.helpers._config_property import ConfigProperty
 from singer_sdk.typing import (
     ArrayType,
     BooleanType,
@@ -19,6 +20,47 @@
 )
 
 _EnumMemberT = t.TypeVar("_EnumMemberT")
+T = t.TypeVar("T")
+
+
+class Builtin:
+    """Use this class to define built-in setting(s) for a plugin."""
+
+    def __init__(
+        self,
+        schema: dict[str, t.Any],
+        *,
+        capability: CapabilitiesEnum | None = None,
+        **kwargs: t.Any,
+    ):
+        """Initialize the descriptor.
+
+        Args:
+            schema: The JSON schema for the setting.
+            capability: The capability that the setting is associated with.
+            kwargs: Additional keyword arguments.
+        """
+        self.schema = schema
+        self.capability = capability
+        self.kwargs = kwargs
+
+    def attribute(  # noqa: PLR6301
+        self,
+        custom_key: str | None = None,
+        *,
+        default: T | None = None,
+    ) -> ConfigProperty[T]:
+        """Generate a class attribute for the setting.
+
+        Args:
+            custom_key: Custom key to use in the config.
+            default: Default value for the setting.
+
+        Returns:
+            Class attribute for the setting.
+        """
+        return ConfigProperty(custom_key=custom_key, default=default)
+
 
 # Default JSON Schema to support config for built-in capabilities:
 
@@ -160,6 +202,16 @@
     ),
 ).to_dict()
 
+TARGET_ALLOW_COLUMN_ALTER_CONFIG = Builtin(
+    schema=PropertiesList(
+        Property(
+            "allow_column_alter",
+            BooleanType,
+            description="Allow altering columns in the target database.",
+        ),
+    ).to_dict(),
+)
+
 
 class TargetLoadMethods(str, Enum):
     """Target-specific capabilities."""

singer_sdk/target_base.py

@@ -18,6 +18,7 @@
 from singer_sdk.helpers.capabilities import (
     ADD_RECORD_METADATA_CONFIG,
     BATCH_CONFIG,
+    TARGET_ALLOW_COLUMN_ALTER_CONFIG,
     TARGET_BATCH_SIZE_ROWS_CONFIG,
     TARGET_HARD_DELETE_CONFIG,
     TARGET_LOAD_METHOD_CONFIG,
@@ -686,6 +687,8 @@ def _merge_missing(source_jsonschema: dict, target_jsonschema: dict) -> None:
                 if k not in target_jsonschema["properties"]:
                     target_jsonschema["properties"][k] = v
 
+        _merge_missing(TARGET_ALLOW_COLUMN_ALTER_CONFIG.schema, config_jsonschema)
+
         capabilities = cls.capabilities
 
         if TargetCapabilities.TARGET_SCHEMA in capabilities:

tests/core/helpers/test_config_property.py

@@ -0,0 +1,33 @@
+"""Test the BuiltinSetting descriptor."""
+
+from __future__ import annotations
+
+from singer_sdk.helpers._config_property import ConfigProperty
+
+
+def test_builtin_setting_descriptor():
+    class ObjWithConfig:
+        example = ConfigProperty(default=1)
+
+        def __init__(self):
+            self.config = {"example": 1}
+
+    obj = ObjWithConfig()
+    assert obj.example == 1
+
+    obj.config["example"] = 2
+    assert obj.example == 2
+
+
+def test_builtin_setting_descriptor_custom_key():
+    class ObjWithConfig:
+        my_attr = ConfigProperty("example", default=1)
+
+        def __init__(self):
+            self.config = {"example": 1}
+
+    obj = ObjWithConfig()
+    assert obj.my_attr == 1
+
+    obj.config["example"] = 2
+    assert obj.my_attr == 2

tests/core/targets/test_target_sql.py

@@ -48,6 +48,7 @@ class MyTarget(SQLTargetMock, capabilities=capabilities):
     about = MyTarget._get_about_info()
     default_settings = {
         "add_record_metadata",
+        "allow_column_alter",
         "load_method",
         "batch_size_rows",
     }

tests/core/test_connector_sql.py

@@ -293,7 +293,12 @@ def test_engine_json_serialization(self, connector: SQLConnector):
 class TestDuckDBConnector:
     @pytest.fixture
     def connector(self):
-        return DuckDBConnector(config={"sqlalchemy_url": "duckdb:///"})
+        return DuckDBConnector(
+            config={
+                "sqlalchemy_url": "duckdb:///",
+                "allow_column_alter": True,
+            },
+        )
 
     def test_create_schema(self, connector: DuckDBConnector):
         engine = connector._engine
@@ -321,7 +326,6 @@ def test_column_rename(self, connector: DuckDBConnector):
             assert result.keys() == ["id", "new_name"]
 
     def test_adapt_column_type(self, connector: DuckDBConnector):
-        connector.allow_column_alter = True
         engine = connector._engine
         meta = sa.MetaData()
         _ = sa.Table(
@@ -341,6 +345,24 @@ def test_adapt_column_type(self, connector: DuckDBConnector):
             assert result.keys() == ["id", "name"]
             assert result.cursor.description[1][1] == "STRING"
 
+    def test_adapt_column_type_not_allowed(self, connector: DuckDBConnector):
+        connector.config["allow_column_alter"] = False
+        engine = connector._engine
+        meta = sa.MetaData()
+        _ = sa.Table(
+            "test_table",
+            meta,
+            sa.Column("id", sa.Integer),
+            sa.Column("name", sa.Integer),
+        )
+        meta.create_all(engine)
+
+        with pytest.raises(
+            NotImplementedError,
+            match="Altering columns is not supported",
+        ):
+            connector._adapt_column_type("test_table", "name", sa.types.String())
+
 
 def test_adapter_without_json_serde():
     registry.register(