Feat: Add Cloud Interop and Robust Secrets Management

.github/workflows/python_pytest.yml

@@ -70,9 +70,11 @@ jobs:
     # Job-specific step(s):
     - name: Run Pytest (No-Creds)
       env:
-        # Force this to an invalid value to ensure tests that no creds are required are run.
-        GCP_GSM_CREDENTIALS: "no-creds"
-      run: poetry run pytest -m "not requires_creds"
+        # Force this to a blank value.
+        GCP_GSM_CREDENTIALS: ""
+      run: >
+        poetry run pytest -m
+        "not requires_creds and not linting and not super_slow"
 
   pytest:
     name: Pytest (All, Python ${{ matrix.python-version }}, ${{ matrix.os }})
@@ -114,4 +116,5 @@ jobs:
     - name: Run Pytest
       env:
         GCP_GSM_CREDENTIALS: ${{ secrets.GCP_GSM_CREDENTIALS }}
-      run: poetry run pytest -m "not linting"
+      run: >
+        poetry run pytest -m "not linting and not super_slow"

.github/workflows/test-pr-command.yml

@@ -82,7 +82,7 @@ jobs:
     - name: Run Pytest
       env:
         GCP_GSM_CREDENTIALS: ${{ secrets.GCP_GSM_CREDENTIALS }}
-      run: poetry run pytest
+      run: poetry run pytest -m "not super_slow"
 
   log-success-comment:
     name: Append 'Success' Comment

.gitignore

@@ -1,6 +1,6 @@
-# Directories and subdirectories called 'secrets' or '.secrets'
+# Directories and subdirectories called '.secrets' and the top-level '/secrets' directory
 .secrets
-secrets
+/secrets
 
 # Virtual Environments
 .venv

README.md

@@ -29,24 +29,34 @@ PyAirbyte can auto-import secrets from the following sources:
 3. [Google Colab secrets](https://medium.com/@parthdasawant/how-to-use-secrets-in-google-colab-450c38e3ec75).
 4. Manual entry via [`getpass`](https://docs.python.org/3.9/library/getpass.html).
 
-_Note: Additional secret store options may be supported in the future. [More info here.](https://github.com/airbytehq/airbyte-lib-private-beta/discussions/5)_
+_Note: You can also build your own secret manager by subclassing the `CustomSecretManager` implementation. For more information, see the `airbyte.secrets.CustomSecretManager` class definiton._
 
 ### Retrieving Secrets
 
 ```python
-from airbyte import get_secret, SecretSource
+import airbyte as ab
 
-source = get_source("source-github")
+source = ab.get_source("source-github")
 source.set_config(
    "credentials": {
-      "personal_access_token": get_secret("GITHUB_PERSONAL_ACCESS_TOKEN"),
+      "personal_access_token": ab.get_secret("GITHUB_PERSONAL_ACCESS_TOKEN"),
    }
 )
 ```
 
-The `get_secret()` function accepts an optional `source` argument of enum type `SecretSource`. If omitted or set to `SecretSource.ANY`, PyAirbyte will search all available secrets sources. If `source` is set to a specific source, then only that source will be checked. If a list of `SecretSource` entries is passed, then the sources will be checked using the provided ordering.
+By default, PyAirbyte will search all available secrets sources. The `get_secret()` function also accepts an optional `sources` argument of specific source names (`SecretSourceEnum`) and/or secret manager objects to check.
 
-By default, PyAirbyte will prompt the user for any requested secrets that are not provided via other secret managers. You can disable this prompt by passing `prompt=False` to `get_secret()`.
+By default, PyAirbyte will prompt the user for any requested secrets that are not provided via other secret managers. You can disable this prompt by passing `allow_prompt=False` to `get_secret()`.
+
+For more information, see the `airbyte.secrets` module.
+
+### Secrets Auto-Discovery
+
+If you have a secret matching an expected name, PyAirbyte will automatically use it. For example, if you have a secret named `GITHUB_PERSONAL_ACCESS_TOKEN`, PyAirbyte will automatically use it when configuring the GitHub source.
+
+The naming convention for secrets is as `{CONNECTOR_NAME}_{PROPERTY_NAME}`, for instance `SNOWFLAKE_PASSWORD` and `BIGQUERY_CREDENTIALS_PATH`.
+
+PyAirbyte will also auto-discover secrets for interop with hosted Airbyte: `AIRBYTE_CLOUD_API_URL`, `AIRBYTE_CLOUD_API_KEY`, etc.
 
 ## Connector compatibility
 
@@ -120,7 +130,6 @@ Yes. Just pick the cache type matching the destination - like SnowflakeCache for
 **6. Can PyAirbyte import a connector from a local directory that has python project files, or does it have to be pip install**
 Yes, PyAirbyte can use any local install that has a CLI - and will automatically find connectors by name if they are on PATH.
 
-
 ## Changelog and Release Notes
 
 For a version history and list of all changes, please see our [GitHub Releases](https://github.com/airbytehq/PyAirbyte/releases) page.

airbyte/__init__.py

@@ -1,3 +1,4 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 """PyAirbyte brings Airbyte ELT to every Python developer.
 
 .. include:: ../README.md
@@ -7,14 +8,14 @@
 """
 from __future__ import annotations
 
-from airbyte import caches, datasets, documents, exceptions, results, secrets, sources
+from airbyte import caches, cloud, datasets, documents, exceptions, results, secrets, sources
 from airbyte.caches.bigquery import BigQueryCache
 from airbyte.caches.duckdb import DuckDBCache
 from airbyte.caches.util import get_default_cache, new_local_cache
 from airbyte.datasets import CachedDataset
 from airbyte.records import StreamRecord
 from airbyte.results import ReadResult
-from airbyte.secrets import SecretSource, get_secret
+from airbyte.secrets import SecretSourceEnum, get_secret
 from airbyte.sources import registry
 from airbyte.sources.base import Source
 from airbyte.sources.registry import get_available_connectors
@@ -23,6 +24,7 @@
 
 __all__ = [
     # Modules
+    "cloud",
     "caches",
     "datasets",
     "documents",
@@ -43,7 +45,7 @@
     "CachedDataset",
     "DuckDBCache",
     "ReadResult",
-    "SecretSource",
+    "SecretSourceEnum",
     "Source",
     "StreamRecord",
 ]

airbyte/_executor.py

@@ -47,7 +47,7 @@ def __init__(
         The 'name' param is required if 'metadata' is None.
         """
         if not name and not metadata:
-            raise exc.AirbyteLibInternalError(message="Either name or metadata must be provided.")
+            raise exc.PyAirbyteInternalError(message="Either name or metadata must be provided.")
 
         self.name: str = name or cast(ConnectorMetadata, metadata).name  # metadata is not None here
         self.metadata: ConnectorMetadata | None = metadata
@@ -270,7 +270,7 @@ def _get_installed_version(
         if not self.interpreter_path.exists():
             # No point in trying to detect the version if the interpreter does not exist
             if raise_on_error:
-                raise exc.AirbyteLibInternalError(
+                raise exc.PyAirbyteInternalError(
                     message="Connector's virtual environment interpreter could not be found.",
                     context={
                         "interpreter_path": self.interpreter_path,

airbyte/_processors/base.py

@@ -60,7 +60,7 @@ def __init__(
         self._expected_streams: set[str] | None = None
         self.cache: CacheBase = cache
         if not isinstance(self.cache, CacheBase):
-            raise exc.AirbyteLibInputError(
+            raise exc.PyAirbyteInputError(
                 message=(
                     f"Expected config class of type 'CacheBase'.  "
                     f"Instead received type '{type(self.cache).__name__}'."
@@ -92,7 +92,7 @@ def register_source(
     ) -> None:
         """Register the source name and catalog."""
         if not self._catalog_manager:
-            raise exc.AirbyteLibInternalError(
+            raise exc.PyAirbyteInternalError(
                 message="Catalog manager should exist but does not.",
             )
         self._catalog_manager.register_source(
@@ -226,7 +226,7 @@ def _finalize_state_messages(
     ) -> None:
         """Handle state messages by passing them to the catalog manager."""
         if not self._catalog_manager:
-            raise exc.AirbyteLibInternalError(
+            raise exc.PyAirbyteInternalError(
                 message="Catalog manager should exist but does not.",
             )
         if state_messages and self._source_name:
@@ -251,7 +251,7 @@ def _get_stream_config(
     ) -> ConfiguredAirbyteStream:
         """Return the definition of the given stream."""
         if not self._catalog_manager:
-            raise exc.AirbyteLibInternalError(
+            raise exc.PyAirbyteInternalError(
                 message="Catalog manager should exist but does not.",
             )
 

airbyte/_processors/file/base.py

@@ -162,7 +162,7 @@ def process_record_message(
             batch_handle = self._new_batch(stream_name=stream_name)
 
         if batch_handle.open_file_writer is None:
-            raise exc.AirbyteLibInternalError(message="Expected open file writer.")
+            raise exc.PyAirbyteInternalError(message="Expected open file writer.")
 
         self._write_record_dict(
             record_dict=StreamRecord.from_record_message(

airbyte/_processors/sql/base.py

@@ -296,7 +296,7 @@ def _get_table_by_name(
         query. To ignore the cache and force a refresh, set 'force_refresh' to True.
         """
         if force_refresh and shallow_okay:
-            raise exc.AirbyteLibInternalError(
+            raise exc.PyAirbyteInternalError(
                 message="Cannot force refresh and use shallow query at the same time."
             )
 
@@ -453,7 +453,7 @@ def _ensure_compatible_table_schema(
         ]
         if missing_columns:
             if raise_on_error:
-                raise exc.AirbyteLibCacheTableValidationError(
+                raise exc.PyAirbyteCacheTableValidationError(
                     violation="Cache table is missing expected columns.",
                     context={
                         "stream_column_names": stream_column_names,
@@ -666,7 +666,7 @@ def _write_files_to_new_table(
 
             # Pandas will auto-create the table if it doesn't exist, which we don't want.
             if not self._table_exists(temp_table_name):
-                raise exc.AirbyteLibInternalError(
+                raise exc.PyAirbyteInternalError(
                     message="Table does not exist after creation.",
                     context={
                         "temp_table_name": temp_table_name,
@@ -727,7 +727,7 @@ def _write_temp_table_to_final_table(
         has_pks: bool = bool(self._get_primary_keys(stream_name))
         has_incremental_key: bool = bool(self._get_incremental_key(stream_name))
         if write_strategy == WriteStrategy.MERGE and not has_pks:
-            raise exc.AirbyteLibInputError(
+            raise exc.PyAirbyteInputError(
                 message="Cannot use merge strategy on a stream with no primary keys.",
                 context={
                     "stream_name": stream_name,
@@ -783,7 +783,7 @@ def _write_temp_table_to_final_table(
             )
             return
 
-        raise exc.AirbyteLibInternalError(
+        raise exc.PyAirbyteInternalError(
             message="Write strategy is not supported.",
             context={
                 "write_strategy": write_strategy,
@@ -843,9 +843,9 @@ def _swap_temp_table_with_final_table(
         Databases that do not support this syntax can override this method.
         """
         if final_table_name is None:
-            raise exc.AirbyteLibInternalError(message="Arg 'final_table_name' cannot be None.")
+            raise exc.PyAirbyteInternalError(message="Arg 'final_table_name' cannot be None.")
         if temp_table_name is None:
-            raise exc.AirbyteLibInternalError(message="Arg 'temp_table_name' cannot be None.")
+            raise exc.PyAirbyteInternalError(message="Arg 'temp_table_name' cannot be None.")
 
         _ = stream_name
         deletion_name = f"{final_table_name}_deleteme"
@@ -909,7 +909,7 @@ def _get_column_by_name(self, table: str | Table, column_name: str) -> Column:
             # Try to get the column in a case-insensitive manner
             return next(col for col in table.c if col.name.lower() == column_name.lower())
         except StopIteration:
-            raise exc.AirbyteLibInternalError(
+            raise exc.PyAirbyteInternalError(
                 message="Could not find matching column.",
                 context={
                     "table": table,

airbyte/_processors/sql/bigquery.py

@@ -175,7 +175,7 @@ def _table_exists(
             return False
 
         except ValueError as ex:
-            raise exc.AirbyteLibInputError(
+            raise exc.PyAirbyteInputError(
                 message="Invalid project name or dataset name.",
                 context={
                     "table_id": table_id,
@@ -225,9 +225,9 @@ def _swap_temp_table_with_final_table(
         ALTER TABLE my_schema.my_old_table_name RENAME TO my_new_table_name;
         """
         if final_table_name is None:
-            raise exc.AirbyteLibInternalError(message="Arg 'final_table_name' cannot be None.")
+            raise exc.PyAirbyteInternalError(message="Arg 'final_table_name' cannot be None.")
         if temp_table_name is None:
-            raise exc.AirbyteLibInternalError(message="Arg 'temp_table_name' cannot be None.")
+            raise exc.PyAirbyteInternalError(message="Arg 'temp_table_name' cannot be None.")
 
         _ = stream_name
         deletion_name = f"{final_table_name}_deleteme"

airbyte/_util/api_duck_types.py

@@ -0,0 +1,21 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+"""A set of duck-typed classes for working with the Airbyte API."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol
+
+
+if TYPE_CHECKING:
+    import requests
+
+
+class AirbyteApiResponseDuckType(Protocol):
+    """Used for duck-typing various Airbyte API responses."""
+
+    content_type: str
+    r"""HTTP response content type for this operation"""
+    status_code: int
+    r"""HTTP response status code for this operation"""
+    raw_response: requests.Response
+    r"""Raw HTTP response; suitable for custom response parsing"""