From 0441fdfb104d93a50df5d6c31b18ba0c9f2988d7 Mon Sep 17 00:00:00 2001
From: "Aaron (\"AJ\") Steers" <aj@airbyte.io>
Date: Tue, 20 Feb 2024 07:41:27 -0800
Subject: [PATCH] Feat: Allow stream selection in `get_source()` and `read()`,
 with support for `streams="*"` (#52)

---
 .github/workflows/python_lint.yml         |  4 ++--
 airbyte/_factories/connector_factories.py |  5 +++++
 airbyte/source.py                         | 23 ++++++++++++++++++++---
 3 files changed, 27 insertions(+), 5 deletions(-)

diff --git a/.github/workflows/python_lint.yml b/.github/workflows/python_lint.yml
index 30e08d3f..66e92d0d 100644
--- a/.github/workflows/python_lint.yml
+++ b/.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 .
diff --git a/airbyte/_factories/connector_factories.py b/airbyte/_factories/connector_factories.py
index f70f2d07..43793284 100644
--- a/airbyte/_factories/connector_factories.py
+++ b/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,
diff --git a/airbyte/source.py b/airbyte/source.py
index 40b69629..e9f5fbd4 100644
--- a/airbyte/source.py
+++ b/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,