From 6768ecca1a66c2d658ef635a42e99440e2c3a348 Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Mon, 19 Feb 2024 15:32:11 -0800
Subject: [PATCH] feat: streamline stream selection

---
 airbyte/_factories/connector_factories.py |  7 ++++++-
 airbyte/source.py                         | 14 +++++++++++---
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/airbyte/_factories/connector_factories.py b/airbyte/_factories/connector_factories.py
index f70f2d07..20f22705 100644
--- a/airbyte/_factories/connector_factories.py
+++ b/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,
diff --git a/airbyte/source.py b/airbyte/source.py
index 40b69629..1861417b 100644
--- a/airbyte/source.py
+++ b/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: