feat: introduce function in Tap to convert input catalog to Stream objects, without discovery.

This clarifies whether the tap should perform discovery or use the input catalog, the two being mutually exclusive.

Author: Rafał Krupiński

cookiecutter/tap-template/{{cookiecutter.tap_id}}/{{cookiecutter.library_name}}/tap.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from singer_sdk import {{ 'SQL' if cookiecutter.stream_type == 'SQL' else '' }}Tap
 from singer_sdk import typing as th  # JSON schema typing helpers
 
@@ -14,6 +16,9 @@
 from {{ cookiecutter.library_name }} import streams
 {%- endif %}
 
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
 
 class Tap{{ cookiecutter.source_name }}({{ 'SQL' if cookiecutter.stream_type == 'SQL' else '' }}Tap):
     """{{ cookiecutter.source_name }} tap class."""
@@ -76,6 +81,18 @@ def discover_streams(self) -> list[streams.{{ cookiecutter.source_name }}Stream]
             streams.GroupsStream(self),
             streams.UsersStream(self),
         ]
+
+    def load_streams_from_catalog(self) -> Iterable[streams.{{cookiecutter.source_name}}Stream]:
+        """Return a list of streams based on the input catalog.
+
+        Returns:
+            A list of requested streams.
+        """
+
+        if "groups" in self.input_catalog:
+            yield streams.GroupsStream(self)
+        if "users" in self.input_catalog:
+            yield streams.UsersStream(self)
 {%- endif %}
 
 

singer_sdk/tap_base.py

@@ -34,6 +34,7 @@
 from singer_sdk.singerlib import Catalog, StateMessage
 
 if t.TYPE_CHECKING:
+    from collections.abc import Iterable
     from pathlib import PurePath
     from types import FrameType
 
@@ -153,13 +154,14 @@ def streams(self) -> dict[str, Stream]:
             A mapping of names to streams, using discovery or a provided catalog.
         """
         if self._streams is None:
-            self._streams = {}
             input_catalog = self.input_catalog
-
-            for stream in self.load_streams():
-                if input_catalog is not None:
+            if input_catalog is None:
+                self._streams = {stream.name: stream for stream in self.load_streams()}
+            else:
+                self._streams = {}
+                for stream in self.load_streams_from_catalog():
                     stream.apply_catalog(input_catalog)
-                self._streams[stream.name] = stream
+                    self._streams[stream.name] = stream
         return self._streams
 
     @property
@@ -408,6 +410,16 @@ def load_streams(self) -> list[Stream]:
             reverse=False,
         )
 
+    def load_streams_from_catalog(self) -> Iterable[Stream]:
+        """Recreate streams from the input catalog.
+
+        For backwards compatibility calls `self.load_streams()`
+
+        Returns:
+            A mapping of names to streams, using provided catalog.
+        """
+        return self.load_streams()
+
     # Bookmarks and state management
 
     def load_state(self, state: dict[str, t.Any]) -> None: