Fix: Updated handling for Manifest-only connectors; Breaks: Remove Python 3.9 support (#340)

Author: aaronsteers

.github/workflows/autofix.yml

@@ -21,7 +21,7 @@ jobs:
     - name: Set up Python
       uses: actions/setup-python@v5
       with:
-        python-version: 3.9
+        python-version: 3.10
         cache: 'poetry'
 
     - name: Install dependencies

.github/workflows/python_pytest.yml

@@ -131,7 +131,6 @@ jobs:
     strategy:
       matrix:
         python-version: [
-          '3.9',
           '3.10',
           '3.11',
         ]

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

@@ -54,7 +54,6 @@ jobs:
     strategy:
       matrix:
         python-version: [
-          '3.9',
           '3.10',
           '3.11',
         ]

airbyte/_executors/declarative.py

@@ -9,12 +9,13 @@
 from typing import IO, TYPE_CHECKING, cast
 
 import pydantic
+import yaml
 
 from airbyte_cdk.entrypoint import AirbyteEntrypoint
 from airbyte_cdk.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
 
+from airbyte import exceptions as exc
 from airbyte._executors.base import Executor
-from airbyte.exceptions import PyAirbyteInternalError
 
 
 if TYPE_CHECKING:
@@ -40,6 +41,7 @@ class DeclarativeExecutor(Executor):
 
     def __init__(
         self,
+        name: str,
         manifest: dict | Path,
     ) -> None:
         """Initialize a declarative executor.
@@ -49,22 +51,36 @@ def __init__(
         - If `manifest` is a dict, it will be used as is.
         """
         _suppress_cdk_pydantic_deprecation_warnings()
+
+        self.name = name
         self._manifest_dict: dict
         if isinstance(manifest, Path):
-            self._manifest_dict = cast(dict, json.loads(manifest.read_text()))
+            self._manifest_dict = cast(dict, yaml.safe_load(manifest.read_text()))
 
         elif isinstance(manifest, dict):
             self._manifest_dict = manifest
 
-        if not isinstance(self._manifest_dict, dict):
-            raise PyAirbyteInternalError(message="Manifest must be a dict.")
-
+        self._validate_manifest(self._manifest_dict)
         self.declarative_source = ManifestDeclarativeSource(source_config=self._manifest_dict)
 
         # TODO: Consider adding version detection
         # https://github.com/airbytehq/airbyte/issues/318
         self.reported_version: str | None = None
 
+    def _validate_manifest(self, manifest_dict: dict) -> None:
+        """Validate the manifest."""
+        manifest_text = json.dumps(manifest_dict)
+        if "class_name:" in manifest_text:
+            raise exc.AirbyteConnectorInstallationError(
+                message=(
+                    "The provided manifest requires additional code files (`class_name` key "
+                    "detected). This feature is not compatible with the declarative YAML "
+                    "executor. To use this executor, please try again with the Python "
+                    "executor."
+                ),
+                connector_name=self.name,
+            )
+
     @property
     def _cli(self) -> list[str]:
         """Not applicable."""

airbyte/_executors/util.py

@@ -4,12 +4,12 @@
 import shutil
 import sys
 import tempfile
-from json import JSONDecodeError
 from pathlib import Path
 from typing import TYPE_CHECKING, cast
 
 import requests
 import yaml
+from requests import HTTPError
 from rich import print
 
 from airbyte import exceptions as exc
@@ -25,6 +25,58 @@
     from airbyte._executors.base import Executor
 
 
+def _try_get_source_manifest(source_name: str, manifest_url: str | None) -> dict:
+    """Try to get a source manifest from a URL.
+
+    If the URL is not provided, we'll try a couple of default URLs.
+    We can remove/refactor this once manifests are available in GCS connector registry.
+    """
+    if manifest_url:
+        response = requests.get(url=manifest_url)
+        response.raise_for_status()  # Raise HTTPError exception if the download failed
+        try:
+            return cast(dict, yaml.safe_load(response.text))
+        except yaml.YAMLError as ex:
+            raise exc.AirbyteConnectorInstallationError(
+                message="Failed to parse the connector manifest YAML.",
+                connector_name=source_name,
+                context={
+                    "manifest_url": manifest_url,
+                },
+            ) from ex
+
+    # No manifest URL was provided. We'll try a couple of default URLs.
+
+    try:
+        # First try the new URL format (language='manifest-only'):
+        result_1 = _try_get_source_manifest(
+            source_name=source_name,
+            manifest_url=(
+                f"https://raw.githubusercontent.com/airbytehq/airbyte/master/airbyte-integrations"
+                f"/connectors/{source_name}/manifest.yaml"
+            ),
+        )
+    except HTTPError as ex_1:
+        # If the new URL path was not found, try the old URL format (language='low-code'):
+        try:
+            result_2 = _try_get_source_manifest(
+                source_name=source_name,
+                manifest_url=(
+                    f"https://raw.githubusercontent.com/airbytehq/airbyte/master/airbyte-integrations"
+                    f"/connectors/{source_name}/{source_name.replace('-', '_')}/manifest.yaml"
+                ),
+            )
+        except HTTPError:
+            # Raise the first exception, since that represents the new default URL
+            raise ex_1 from None
+        else:
+            # Old URL path was found (no exceptions raised).
+            return result_2
+    else:
+        # New URL path was found (no exceptions raised).
+        return result_1
+
+
 def get_connector_executor(  # noqa: PLR0912, PLR0913, PLR0915 # Too complex
     name: str,
     *,
@@ -143,49 +195,23 @@ def get_connector_executor(  # noqa: PLR0912, PLR0913, PLR0915 # Too complex
         )
 
     if source_manifest:
-        if source_manifest is True:
-            # Auto-set the manifest to a valid http address URL string
-            source_manifest = (
-                "https://raw.githubusercontent.com/airbytehq/airbyte/master/airbyte-integrations"
-                f"/connectors/{name}/{name.replace('-', '_')}/manifest.yaml"
+        if isinstance(source_manifest, (dict, Path)):
+            return DeclarativeExecutor(
+                name=name,
+                manifest=source_manifest,
             )
-        if isinstance(source_manifest, str):
-            print("Installing connector from YAML manifest:", source_manifest)
-            # Download the manifest file
-            response = requests.get(url=source_manifest)
-            response.raise_for_status()  # Raise an exception if the download failed
-
-            if "class_name:" in response.text:
-                raise exc.AirbyteConnectorInstallationError(
-                    message=(
-                        "The provided manifest requires additional code files (`class_name` key "
-                        "detected). This feature is not compatible with the declarative YAML "
-                        "executor. To use this executor, please try again with the Python "
-                        "executor."
-                    ),
-                    connector_name=name,
-                    context={
-                        "manifest_url": source_manifest,
-                    },
-                )
 
-            try:
-                source_manifest = cast(dict, yaml.safe_load(response.text))
-            except JSONDecodeError as ex:
-                raise exc.AirbyteConnectorInstallationError(
-                    connector_name=name,
-                    context={
-                        "manifest_url": source_manifest,
-                    },
-                ) from ex
-
-        if isinstance(source_manifest, Path):
-            source_manifest = cast(dict, yaml.safe_load(source_manifest.read_text()))
-
-        # Source manifest is a dict at this point
-        return DeclarativeExecutor(
-            manifest=source_manifest,
-        )
+        if isinstance(source_manifest, (str, bool)):
+            # Source manifest is either a URL or a boolean (True)
+            source_manifest = _try_get_source_manifest(
+                source_name=name,
+                manifest_url=None if source_manifest is True else source_manifest,
+            )
+
+            return DeclarativeExecutor(
+                name=name,
+                manifest=source_manifest,
+            )
 
     # else: we are installing a connector in a Python virtual environment:
 

airbyte/secrets/__init__.py

@@ -11,7 +11,7 @@
 1. Environment variables.
 2. Variables defined in a local `.env` ("Dotenv") file.
 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).
+4. Manual entry via [`getpass`](https://docs.python.org/3.10/library/getpass.html).
 
 **Note:** You can also build your own secret manager by subclassing the `CustomSecretManager`
 implementation. For more information, see the `airbyte.secrets.CustomSecretManager` reference docs.

airbyte/sources/registry.py

@@ -24,6 +24,7 @@
 _REGISTRY_URL = "https://connectors.airbyte.com/files/registries/v0/oss_registry.json"
 
 _LOWCODE_LABEL = "cdk:low-code"
+_MANIFEST_ONLY_LABEL = "cdk:manifest-only"
 
 _LOWCODE_CONNECTORS_NEEDING_PYTHON: list[str] = [
     "source-adjust",
@@ -169,7 +170,10 @@ def _registry_entry_to_connector_metadata(entry: dict) -> ConnectorMetadata:
             InstallType.DOCKER if entry.get("dockerImageTag") else None,
             InstallType.PYTHON if pypi_enabled else None,
             InstallType.JAVA if language == Language.JAVA else None,
-            InstallType.YAML if _LOWCODE_LABEL in entry.get("tags", []) else None,
+            InstallType.YAML
+            if _LOWCODE_LABEL in entry.get("tags", [])
+            or _MANIFEST_ONLY_LABEL in entry.get("tags", [])
+            else None,
         ]
         if x
     }

examples/run_sync_to_destination_from_read_result.py

@@ -20,13 +20,11 @@
 def get_my_source() -> ab.Source:
     return ab.get_source(
         "source-faker",
-        local_executable="source-faker",
         config={
             "count": SCALE,
             "seed": 1234,
             "parallelism": 16,
         },
-        install_if_missing=False,
         streams=["purchases"],
     )
 

examples/run_sync_to_destination_w_cache.py

@@ -20,13 +20,11 @@
 def get_my_source() -> ab.Source:
     return ab.get_source(
         "source-faker",
-        local_executable="source-faker",
         config={
             "count": SCALE,
             "seed": 1234,
             "parallelism": 16,
         },
-        install_if_missing=False,
         streams="*",
     )
 

examples/run_sync_to_destination_wo_cache.py

@@ -20,13 +20,11 @@
 def get_my_source() -> ab.Source:
     return ab.get_source(
         "source-faker",
-        local_executable="source-faker",
         config={
             "count": SCALE,
             "seed": 1234,
             "parallelism": 16,
         },
-        install_if_missing=False,
         streams=["purchases"],
     )