feat: Support deselecting streams by default (#1672)

* feat: Support deselecting streams by default

* Add tests

singer_sdk/_singerlib/catalog.py

@@ -166,6 +166,7 @@ def get_standard_metadata(
         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 @@ def get_standard_metadata(
             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 @@ def get_standard_metadata(
             table_key_properties=key_properties,
             forced_replication_method=replication_method,
             valid_replication_keys=valid_replication_keys,
+            selected_by_default=selected_by_default,
         )
 
         if schema:

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 @@ def primary_keys(self) -> list[str] | None:
         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 @@ def replication_key(self) -> str | None:
         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 @@ def metadata(self) -> singer.MetadataMapping:
                 [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
 

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

tests/core/test_streams.py

@@ -516,3 +516,81 @@ def calculate_test_cost(
     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]