refactor: Use a single source of truth for built-in capabilities

docs/builtin.rst

@@ -0,0 +1,36 @@
+Built-in Settings and Capabilities
+==================================
+
+.. currentmodule:: singer_sdk.helpers.capabilities
+
+The Singer SDK library provides a number of built-in settings and capabilities.
+
+.. autodata:: ACTIVATE_VERSION
+   :no-value:
+
+.. autodata:: ADD_RECORD_METADATA
+   :no-value:
+
+.. autodata:: BATCH
+   :no-value:
+
+.. autodata:: FLATTENING
+   :no-value:
+
+.. autodata:: STREAM_MAPS
+   :no-value:
+
+.. autodata:: TARGET_BATCH_SIZE_ROWS
+   :no-value:
+
+.. autodata:: TARGET_HARD_DELETE
+   :no-value:
+
+.. autodata:: TARGET_LOAD_METHOD
+   :no-value:
+
+.. autodata:: TARGET_SCHEMA
+   :no-value:
+
+.. autodata:: TARGET_VALIDATE_RECORDS
+   :no-value:

docs/index.rst

@@ -57,6 +57,7 @@ within the `#singer-tap-development`_ and `#singer-target-development`_ Slack ch
    implementation/index
    typing
    capabilities
+   builtin
 
 .. toctree::
    :caption: Advanced Concepts

singer_sdk/connectors/sql.py

@@ -17,8 +17,8 @@
 from singer_sdk import typing as th
 from singer_sdk._singerlib import CatalogEntry, MetadataMapping, Schema
 from singer_sdk.exceptions import ConfigValidationError
+from singer_sdk.helpers import capabilities
 from singer_sdk.helpers._util import dump_json, load_json
-from singer_sdk.helpers.capabilities import TargetLoadMethods
 
 if sys.version_info < (3, 13):
     from typing_extensions import deprecated
@@ -1247,7 +1247,7 @@ def prepare_table(
                 as_temp_table=as_temp_table,
             )
             return
-        if self.config["load_method"] == TargetLoadMethods.OVERWRITE:
+        if self.config["load_method"] == capabilities.TargetLoadMethods.OVERWRITE:
             self.get_table(full_table_name=full_table_name).drop(self._engine)
             self.create_empty_table(
                 full_table_name=full_table_name,

singer_sdk/helpers/capabilities/__init__.py

@@ -0,0 +1,191 @@
+"""Module with helpers to declare capabilities and plugin behavior."""
+
+from __future__ import annotations
+
+from singer_sdk.helpers.capabilities import _schema as schema
+from singer_sdk.helpers.capabilities._builtin import Builtin
+from singer_sdk.helpers.capabilities._config_property import ConfigProperty
+from singer_sdk.helpers.capabilities._enum import (
+    CapabilitiesEnum,
+    PluginCapabilities,
+    TapCapabilities,
+    TargetCapabilities,
+    TargetLoadMethods,
+)
+
+__all__ = [
+    "ADD_RECORD_METADATA",
+    "BATCH",
+    "FLATTENING",
+    "STREAM_MAPS",
+    "TARGET_BATCH_SIZE_ROWS",
+    "TARGET_HARD_DELETE",
+    "TARGET_LOAD_METHOD",
+    "TARGET_SCHEMA",
+    "TARGET_VALIDATE_RECORDS",
+    "CapabilitiesEnum",
+    "ConfigProperty",
+    "PluginCapabilities",
+    "TapCapabilities",
+    "TargetCapabilities",
+    "TargetLoadMethods",
+]
+
+#: Support the `ACTIVATE_VERSION <https://hub.meltano.com/singer/docs#activate-version>`_
+#: extension.
+#:
+#: Example:
+#:
+#: .. code-block:: json
+#:
+#:    {
+#:        "activate_version": true
+#:    }
+#:
+ACTIVATE_VERSION = Builtin(
+    schema=schema.ACTIVATE_VERSION_CONFIG,
+    capability=PluginCapabilities.ACTIVATE_VERSION,
+)
+
+#: Add metadata to records.
+#:
+#: Example:
+#:
+#: .. code-block:: json
+#:
+#:    {
+#:        "add_record_metadata": true
+#:    }
+#:
+ADD_RECORD_METADATA = Builtin(schema=schema.ADD_RECORD_METADATA_CONFIG)
+
+#: For taps, support emitting BATCH messages. For targets, support consuming BATCH
+#: messages.
+#:
+#: Example:
+#:
+#: .. code-block:: json
+#:
+#:    {
+#:        "batch_config": {
+#:            "encoding": {
+#:                "format": "jsonl",
+#:                "compression": "gzip"
+#:            },
+#:            "storage": {
+#:                "type": "root",
+#:                "root": "file:///path/to/batch/files",
+#:                "prefix": "batch-"
+#:            }
+#:        }
+#:    }
+#:
+BATCH = Builtin(
+    schema=schema.BATCH_CONFIG,
+    capability=PluginCapabilities.BATCH,
+)
+
+#: Support schema flattening, aka de-nesting of complex properties.
+#:
+#: Example:
+#:
+#: .. code-block:: json
+#:
+#:    {
+#:        "flattening_enabled": true,
+#:        "flattening_max_depth": 3
+#:    }
+#:
+FLATTENING = Builtin(
+    schema=schema.FLATTENING_CONFIG,
+    capability=PluginCapabilities.FLATTENING,
+)
+
+#: Support inline stream map transforms.
+#:
+#: Example:
+#:
+#: .. code-block:: json
+#:
+#:    {
+#:      "stream_maps": {
+#:        "users": {
+#:          "id": "id",
+#:          "fields": "[f for f in fields if f['key'] != 'age']"
+#:        }
+#:      }
+#:    }
+#:
+STREAM_MAPS = Builtin(
+    schema.STREAM_MAPS_CONFIG,
+    capability=PluginCapabilities.STREAM_MAPS,
+)
+
+#: Target batch size in rows.
+#:
+#: Example:
+#:
+#: .. code-block:: json
+#:
+#:    {
+#:        "batch_size_rows": 10000
+#:    }
+#:
+TARGET_BATCH_SIZE_ROWS = Builtin(schema=schema.TARGET_BATCH_SIZE_ROWS_CONFIG)
+
+#: Support hard delete capability.
+#:
+#: Example:
+#:
+#: .. code-block:: json
+#:
+#:    {
+#:        "hard_delete": true
+#:    }
+#:
+TARGET_HARD_DELETE = Builtin(
+    schema=schema.TARGET_HARD_DELETE_CONFIG,
+    capability=TargetCapabilities.HARD_DELETE,
+)
+
+#: Target load method.
+#:
+#: Example:
+#:
+#: .. code-block:: json
+#:
+#:    {
+#:        "load_method": "upsert"
+#:    }
+#:
+TARGET_LOAD_METHOD = Builtin(schema=schema.TARGET_LOAD_METHOD_CONFIG)
+
+#: Allow setting the target schema.
+#:
+#: Example:
+#:
+#: .. code-block:: json
+#:
+#:    {
+#:        "default_target_schema": "my_schema"
+#:    }
+#:
+TARGET_SCHEMA = Builtin(
+    schema=schema.TARGET_SCHEMA_CONFIG,
+    capability=TargetCapabilities.TARGET_SCHEMA,
+)
+
+#: Validate incoming records against their declared schema.
+#:
+#: Example:
+#:
+#: .. code-block:: json
+#:
+#:    {
+#:        "validate_records": true
+#:    }
+#:
+TARGET_VALIDATE_RECORDS = Builtin(
+    schema=schema.TARGET_VALIDATE_RECORDS_CONFIG,
+    capability=TargetCapabilities.VALIDATE_RECORDS,
+)

singer_sdk/helpers/capabilities/_builtin.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import typing as t
+
+from ._config_property import ConfigProperty
+
+if t.TYPE_CHECKING:
+    from ._enum import CapabilitiesEnum
+
+_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)

singer_sdk/helpers/capabilities/_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]