Feat: Allow stream selection in get_source() and read(), with sup…

…port for `streams="*"` (#52)

.github/workflows/python_lint.yml

@@ -67,5 +67,5 @@ jobs:
         cache: 'poetry'
 
     # Job-specifc step(s):
-    - name: Check code format
-      run: poetry run ruff format --check .
+    - name: Check MyPy typing
+      run: poetry run mypy .

airbyte/_factories/connector_factories.py

@@ -42,6 +42,7 @@ def get_source(
     name: str,
     config: dict[str, Any] | None = None,
     *,
+    streams: str | list[str] | None = None,
     version: str | None = None,
     pip_url: str | None = None,
     local_executable: Path | str | None = None,
@@ -53,6 +54,9 @@ def get_source(
         name: connector name
         config: connector config - if not provided, you need to set it later via the set_config
             method.
+        streams: list of stream names to select for reading. If set to "*", all streams will be
+            selected. If not provided, you can set it later via the `select_streams()` or
+            `select_all_streams()` method.
         version: connector version - if not provided, the currently installed version will be used.
             If no version is installed, the latest available version will be used. The version can
             also be set to "latest" to force the use of the latest available version.
@@ -88,6 +92,7 @@ def get_source(
         return Source(
             name=name,
             config=config,
+            streams=streams,
             executor=PathExecutor(
                 name=name,
                 path=local_executable,

airbyte/source.py

@@ -76,7 +76,7 @@ def __init__(
         executor: Executor,
         name: str,
         config: dict[str, Any] | None = None,
-        streams: list[str] | None = None,
+        streams: str | list[str] | None = None,
         *,
         validate: bool = False,
     ) -> None:
@@ -95,7 +95,7 @@ def __init__(
         if config is not None:
             self.set_config(config, validate=validate)
         if streams is not None:
-            self.set_streams(streams)
+            self.select_streams(streams)
 
     def set_streams(self, streams: list[str]) -> None:
         """Deprecated. See select_streams()."""
@@ -115,11 +115,22 @@ def select_all_streams(self) -> None:
         """
         self._selected_stream_names = self.get_available_streams()
 
-    def select_streams(self, streams: list[str]) -> None:
+    def select_streams(self, streams: str | list[str]) -> None:
         """Select the stream names that should be read from the connector.
 
+        Args:
+        - streams: A list of stream names to select. If set to "*", all streams will be selected.
+
         Currently, if this is not set, all streams will be read.
         """
+        if streams == "*":
+            self.select_all_streams()
+            return
+
+        if isinstance(streams, str):
+            # If a single stream is provided, convert it to a one-item list
+            streams = [streams]
+
         available_streams = self.get_available_streams()
         for stream in streams:
             if stream not in available_streams:
@@ -513,6 +524,7 @@ def read(
         self,
         cache: SQLCacheBase | None = None,
         *,
+        streams: str | list[str] | None = None,
         write_strategy: str | WriteStrategy = WriteStrategy.AUTO,
         force_full_refresh: bool = False,
     ) -> ReadResult:
@@ -524,6 +536,8 @@ def read(
                 one of "append", "upsert", "replace", or "auto". If a WriteStrategy, it must be one
                 of WriteStrategy.APPEND, WriteStrategy.UPSERT, WriteStrategy.REPLACE, or
                 WriteStrategy.AUTO.
+            streams: Optional if already set. A list of stream names to select for reading. If set
+                to "*", all streams will be selected.
             force_full_refresh: If True, the source will operate in full refresh mode. Otherwise,
                 streams will be read in incremental mode if supported by the connector. This option
                 must be True when using the "replace" strategy.
@@ -551,6 +565,9 @@ def read(
                     },
                 ) from None
 
+        if streams:
+            self.select_streams(streams)
+
         if not self._selected_stream_names:
             raise exc.AirbyteLibNoStreamsSelectedError(
                 connector_name=self.name,