Repository: meltano/sdk (python)

Issue:
[Feature]: Support de-selecting a stream by default
### Feature scope

Taps (catalog, state, stream maps, etc.)

### Description

Tap developers may want to disable certain streams in their taps for a few reasons:

- The stream requires higher access credentials than the rest
- The stream requires a different payment tier in the source system
- The stream has a large volume of data and not all users may be interested in it by default

The discovered catalog would flag the stream with `"selected_by_default": false`. This means that with no catalog file, the stream selection will be resolved to `False`.

End users could still toggle the stream selection as normal by passing a catalog file.

Related discussion:

- https://github.com/meltano/sdk/discussions/1652

Pull request:
feat: Support deselecting streams by default
* Closes #1203 

TODO:

- [x] Tests
- [x] Docs

Patch:
diff --git a/singer_sdk/_singerlib/catalog.py b/singer_sdk/_singerlib/catalog.py
index 5932bb3f..a7284705 100644
--- a/singer_sdk/_singerlib/catalog.py
+++ b/singer_sdk/_singerlib/catalog.py
@@ -166,6 +166,7 @@ class MetadataMapping(t.Dict[Breadcrumb, AnyMetadata]):
         key_properties: list[str] | None = None,
         valid_replication_keys: list[str] | None = None,
         replication_method: str | None = None,
+        selected_by_default: bool | None = None,
     ) -> MetadataMapping:
         """Get default metadata for a stream.
 
@@ -175,6 +176,7 @@ class MetadataMapping(t.Dict[Breadcrumb, AnyMetadata]):
             key_properties: Stream key properties.
             valid_replication_keys: Stream valid replication keys.
             replication_method: Stream replication method.
+            selected_by_default: Whether the stream is selected by default.
 
         Returns:
             Metadata mapping.
@@ -184,6 +186,7 @@ class MetadataMapping(t.Dict[Breadcrumb, AnyMetadata]):
             table_key_properties=key_properties,
             forced_replication_method=replication_method,
             valid_replication_keys=valid_replication_keys,
+            selected_by_default=selected_by_default,
         )
 
         if schema:
diff --git a/singer_sdk/streams/core.py b/singer_sdk/streams/core.py
index e6cfeefc..30f64f9e 100644
--- a/singer_sdk/streams/core.py
+++ b/singer_sdk/streams/core.py
@@ -100,6 +100,9 @@ class Stream(metaclass=abc.ABCMeta):
     # Internal API cost aggregator
     _sync_costs: dict[str, int] = {}
 
+    selected_by_default: bool = True
+    """Whether this stream is selected by default in the catalog."""
+
     def __init__(
         self,
         tap: Tap,
@@ -428,7 +431,7 @@ class Stream(metaclass=abc.ABCMeta):
         return self._primary_keys
 
     @primary_keys.setter
-    def primary_keys(self, new_value: list[str]) -> None:
+    def primary_keys(self, new_value: list[str] | None) -> None:
         """Set primary key(s) for the stream.
 
         Args:
@@ -472,7 +475,7 @@ class Stream(metaclass=abc.ABCMeta):
         return self._replication_key
 
     @replication_key.setter
-    def replication_key(self, new_value: str) -> None:
+    def replication_key(self, new_value: str | None) -> None:
         """Set replication key for the stream.
 
         Args:
@@ -532,11 +535,13 @@ class Stream(metaclass=abc.ABCMeta):
                 [self.replication_key] if self.replication_key else None
             ),
             schema_name=None,
+            selected_by_default=self.selected_by_default,
         )
 
         # If there's no input catalog, select all streams
-        if self._tap_input_catalog is None:
-            self._metadata.root.selected = True
+        self._metadata.root.selected = (
+            self._tap_input_catalog is None and self.selected_by_default
+        )
 
         return self._metadata
 


Test patch:
diff --git a/tests/_singerlib/test_catalog.py b/tests/_singerlib/test_catalog.py
index c589d183..69f37713 100644
--- a/tests/_singerlib/test_catalog.py
+++ b/tests/_singerlib/test_catalog.py
@@ -241,7 +241,7 @@ def test_catalog_parsing():
 )
 def test_standard_metadata(
     schema: dict,
-    key_properties: list[str] | None,
+    key_properties: list[str],
     replication_method: str | None,
     valid_replication_keys: list[str] | None,
     schema_name: str | None,
@@ -256,6 +256,7 @@ def test_standard_metadata(
     )
 
     stream_metadata = metadata[()]
+    assert isinstance(stream_metadata, StreamMetadata)
     assert stream_metadata.table_key_properties == key_properties
     assert stream_metadata.forced_replication_method == replication_method
     assert stream_metadata.valid_replication_keys == valid_replication_keys
diff --git a/tests/core/test_streams.py b/tests/core/test_streams.py
index 21794f0b..71789c1f 100644
--- a/tests/core/test_streams.py
+++ b/tests/core/test_streams.py
@@ -516,3 +516,81 @@ def test_sync_costs_calculation(tap: SimpleTestTap, caplog):
     for record in caplog.records:
         assert record.levelname == "INFO"
         assert f"Total Sync costs for stream {stream.name}" in record.message
+
+
+@pytest.mark.parametrize(
+    "input_catalog,selection",
+    [
+        pytest.param(
+            None,
+            {
+                "selected_stream": True,
+                "unselected_stream": False,
+            },
+            id="no_catalog",
+        ),
+        pytest.param(
+            {
+                "streams": [],
+            },
+            {
+                "selected_stream": False,
+                "unselected_stream": False,
+            },
+            id="empty_catalog",
+        ),
+        pytest.param(
+            {
+                "streams": [
+                    {
+                        "tap_stream_id": "selected_stream",
+                        "metadata": [
+                            {
+                                "breadcrumb": [],
+                                "metadata": {
+                                    "selected": True,
+                                },
+                            },
+                        ],
+                    },
+                    {
+                        "tap_stream_id": "unselected_stream",
+                        "metadata": [
+                            {
+                                "breadcrumb": [],
+                                "metadata": {
+                                    "selected": True,
+                                },
+                            },
+                        ],
+                    },
+                ],
+            },
+            {
+                "selected_stream": True,
+                "unselected_stream": True,
+            },
+            id="catalog_with_selection",
+        ),
+    ],
+)
+def test_stream_class_selection(input_catalog, selection):
+    """Test stream class selection."""
+
+    class SelectedStream(RESTStream):
+        name = "selected_stream"
+        url_base = "https://example.com"
+        schema = {"type": "object", "properties": {}}
+
+    class UnselectedStream(SelectedStream):
+        name = "unselected_stream"
+        selected_by_default = False
+
+    class MyTap(SimpleTestTap):
+        def discover_streams(self):
+            return [SelectedStream(self), UnselectedStream(self)]
+
+    # Check that the selected stream is selected
+    tap = MyTap(config=None, catalog=input_catalog)
+    for stream in selection:
+        assert tap.streams[stream].selected is selection[stream]