fix: Append batch config if target supports the batch capability

singer_sdk/target_base.py

@@ -17,6 +17,7 @@
 from singer_sdk.helpers._classproperty import classproperty
 from singer_sdk.helpers._compat import final
 from singer_sdk.helpers.capabilities import (
+    BATCH_CONFIG,
     TARGET_SCHEMA_CONFIG,
     CapabilitiesEnum,
     PluginCapabilities,
@@ -571,6 +572,38 @@ def get_singer_command(cls: type[Target]) -> click.Command:
 
         return command
 
+    @classmethod
+    def append_builtin_config(cls: type[Target], config_jsonschema: dict) -> None:
+        """Appends built-in config to `config_jsonschema` if not already set.
+
+        To customize or disable this behavior, developers may either override this class
+        method or override the `capabilities` property to disabled any unwanted
+        built-in capabilities.
+
+        For all except very advanced use cases, we recommend leaving these
+        implementations "as-is", since this provides the most choice to users and is
+        the most "future proof" in terms of taking advantage of built-in capabilities
+        which may be added in the future.
+
+        Args:
+            config_jsonschema: [description]
+        """
+
+        def _merge_missing(source_jsonschema: dict, target_jsonschema: dict) -> None:
+            # Append any missing properties in the target with those from source.
+            for k, v in source_jsonschema["properties"].items():
+                if k not in target_jsonschema["properties"]:
+                    target_jsonschema["properties"][k] = v
+
+        capabilities = cls.capabilities
+
+        if PluginCapabilities.BATCH in capabilities:
+            _merge_missing(BATCH_CONFIG, config_jsonschema)
+
+        super().append_builtin_config(config_jsonschema)
+
+    pass
+
 
 class SQLTarget(Target):
     """Target implementation for SQL destinations."""

tests/conftest.py

@@ -12,12 +12,16 @@
 
 from singer_sdk import SQLConnector
 from singer_sdk import typing as th
+from singer_sdk.helpers.capabilities import PluginCapabilities
 from singer_sdk.sinks import BatchSink, SQLSink
 from singer_sdk.target_base import SQLTarget, Target
 
 if t.TYPE_CHECKING:
     from _pytest.config import Config
 
+    from singer_sdk.helpers.capabilities import CapabilitiesEnum
+
+
 SYSTEMS = {"linux", "darwin", "windows"}
 
 pytest_plugins = ("singer_sdk.testing.pytest_plugin",)
@@ -104,6 +108,10 @@ class TargetMock(Target):
     name = "target-mock"
     config_jsonschema = th.PropertiesList().to_dict()
     default_sink_class = BatchSinkMock
+    capabilities: t.ClassVar[list[CapabilitiesEnum]] = [
+        *Target.capabilities,
+        PluginCapabilities.BATCH,
+    ]
 
     def __init__(self, *args, **kwargs):
         """Create the Mock target sync."""

tests/core/test_target_base.py

@@ -8,6 +8,7 @@
     MissingKeyPropertiesError,
     RecordsWithoutSchemaException,
 )
+from singer_sdk.helpers.capabilities import PluginCapabilities
 from tests.conftest import BatchSinkMock, SQLSinkMock, SQLTargetMock, TargetMock
 
 
@@ -58,6 +59,24 @@ def test_validate_record():
         sink._singer_validate_message({"name": "test"})
 
 
+def test_target_about_info():
+    target = TargetMock()
+    about = target._get_about_info()
+
+    assert about.capabilities == [
+        PluginCapabilities.ABOUT,
+        PluginCapabilities.STREAM_MAPS,
+        PluginCapabilities.FLATTENING,
+        PluginCapabilities.BATCH,
+    ]
+
+    assert "stream_maps" in about.settings["properties"]
+    assert "stream_map_config" in about.settings["properties"]
+    assert "flattening_enabled" in about.settings["properties"]
+    assert "flattening_max_depth" in about.settings["properties"]
+    assert "batch_config" in about.settings["properties"]
+
+
 def test_sql_get_sink():
     input_schema_1 = {
         "properties": {