feat: streamline stream selection

airbyte/_factories/connector_factories.py

@@ -4,7 +4,7 @@
 import shutil
 import warnings
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 
 from airbyte import exceptions as exc
 from airbyte._executor import PathExecutor, VenvExecutor
@@ -42,6 +42,7 @@ def get_source(
     name: str,
     config: dict[str, Any] | None = None,
     *,
+    streams: Literal["*"] | 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

@@ -5,7 +5,7 @@
 import tempfile
 import warnings
 from contextlib import contextmanager, suppress
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal
 
 import jsonschema
 import pendulum
@@ -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,19 @@ 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: Literal["*"] | list[str]) -> None:
         """Select the stream names that should be read from the connector.
 
         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: