Merge remote-tracking branch 'origin/main' into aj/feat/cli/add-docker-mode-fast-test-option (manual merge resolution)

Author: aaronsteers

airbyte_cdk/cli/airbyte_cdk/_secrets.py

@@ -43,6 +43,7 @@
 from rich.console import Console
 from rich.table import Table
 
+from airbyte_cdk.cli.airbyte_cdk.exceptions import ConnectorSecretWithNoValidVersionsError
 from airbyte_cdk.utils.connector_paths import (
     resolve_connector_name,
     resolve_connector_name_and_directory,
@@ -131,24 +132,46 @@ def fetch(
     )
     # Fetch and write secrets
     secret_count = 0
+    exceptions = []
+
     for secret in secrets:
         secret_file_path = _get_secret_filepath(
             secrets_dir=secrets_dir,
             secret=secret,
         )
-        _write_secret_file(
-            secret=secret,
-            client=client,
-            file_path=secret_file_path,
+        try:
+            _write_secret_file(
+                secret=secret,
+                client=client,
+                file_path=secret_file_path,
+                connector_name=connector_name,
+                gcp_project_id=gcp_project_id,
+            )
+            click.echo(f"Secret written to: {secret_file_path.absolute()!s}", err=True)
+            secret_count += 1
+        except ConnectorSecretWithNoValidVersionsError as e:
+            exceptions.append(e)
+            click.echo(
+                f"Failed to retrieve secret '{e.secret_name}': No enabled version found", err=True
+            )
+
+    if secret_count == 0 and not exceptions:
+        click.echo(
+            f"No secrets found for connector: '{connector_name}'",
+            err=True,
         )
-        click.echo(f"Secret written to: {secret_file_path.absolute()!s}", err=True)
-        secret_count += 1
 
-    if secret_count == 0:
+    if exceptions:
+        error_message = f"Failed to retrieve {len(exceptions)} secret(s)"
         click.echo(
-            f"No secrets found for connector: '{connector_name}'",
+            style(
+                error_message,
+                fg="red",
+            ),
             err=True,
         )
+        if secret_count == 0:
+            raise exceptions[0]
 
     if not print_ci_secrets_masks:
         return
@@ -230,9 +253,8 @@ def list_(
     table.add_column("Created", justify="left", style="blue", overflow="fold")
     for secret in secrets:
         full_secret_name = secret.name
-        secret_name = full_secret_name.split("/secrets/")[-1]  # Removes project prefix
-        # E.g. https://console.cloud.google.com/security/secret-manager/secret/SECRET_SOURCE-SHOPIFY__CREDS/versions?hl=en&project=<gcp_project_id>
-        secret_url = f"https://console.cloud.google.com/security/secret-manager/secret/{secret_name}/versions?hl=en&project={gcp_project_id}"
+        secret_name = _extract_secret_name(full_secret_name)
+        secret_url = _get_secret_url(secret_name, gcp_project_id)
         table.add_row(
             f"[link={secret_url}]{secret_name}[/link]",
             "\n".join([f"{k}={v}" for k, v in secret.labels.items()]),
@@ -242,6 +264,43 @@ def list_(
     console.print(table)
 
 
+def _extract_secret_name(secret_name: str) -> str:
+    """Extract the secret name from a fully qualified secret path.
+
+    Handles different formats of secret names:
+    - Full path: "projects/project-id/secrets/SECRET_NAME"
+    - Already extracted: "SECRET_NAME"
+
+    Args:
+        secret_name: The secret name or path
+
+    Returns:
+        str: The extracted secret name without project prefix
+    """
+    if "/secrets/" in secret_name:
+        return secret_name.split("/secrets/")[-1]
+    return secret_name
+
+
+def _get_secret_url(secret_name: str, gcp_project_id: str) -> str:
+    """Generate a URL for a secret in the GCP Secret Manager console.
+
+    Note: This URL itself does not contain secrets or sensitive information.
+    The URL itself is only useful for valid logged-in users of the project, and it
+    safe to print this URL in logs.
+
+    Args:
+        secret_name: The name of the secret in GCP.
+        gcp_project_id: The GCP project ID.
+
+    Returns:
+        str: URL to the secret in the GCP console
+    """
+    # Ensure we have just the secret name without the project prefix
+    secret_name = _extract_secret_name(secret_name)
+    return f"https://console.cloud.google.com/security/secret-manager/secret/{secret_name}/versions?hl=en&project={gcp_project_id}"
+
+
 def _fetch_secret_handles(
     connector_name: str,
     gcp_project_id: str = AIRBYTE_INTERNAL_GCP_PROJECT,
@@ -272,9 +331,44 @@ def _write_secret_file(
     secret: "Secret",  # type: ignore
     client: "secretmanager.SecretManagerServiceClient",  # type: ignore
     file_path: Path,
+    connector_name: str,
+    gcp_project_id: str,
 ) -> None:
-    version_name = f"{secret.name}/versions/latest"
-    response = client.access_secret_version(name=version_name)
+    """Write the most recent enabled version of a secret to a file.
+
+    Lists all enabled versions of the secret and selects the most recent one.
+    Raises ConnectorSecretWithNoValidVersionsError if no enabled versions are found.
+
+    Args:
+        secret: The secret to write to a file
+        client: The Secret Manager client
+        file_path: The path to write the secret to
+        connector_name: The name of the connector
+        gcp_project_id: The GCP project ID
+
+    Raises:
+        ConnectorSecretWithNoValidVersionsError: If no enabled version is found
+    """
+    # List all enabled versions of the secret.
+    response = client.list_secret_versions(
+        request={"parent": secret.name, "filter": "state:ENABLED"}
+    )
+
+    # The API returns versions pre-sorted in descending order, with the
+    # 0th item being the latest version.
+    versions = list(response)
+
+    if not versions:
+        secret_name = _extract_secret_name(secret.name)
+        raise ConnectorSecretWithNoValidVersionsError(
+            connector_name=connector_name,
+            secret_name=secret_name,
+            gcp_project_id=gcp_project_id,
+        )
+
+    enabled_version = versions[0]
+
+    response = client.access_secret_version(name=enabled_version.name)
     file_path.write_text(response.payload.data.decode("UTF-8"))
     file_path.chmod(0o600)  # default to owner read/write only
 

airbyte_cdk/cli/airbyte_cdk/exceptions.py

@@ -0,0 +1,23 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+"""Exceptions for the Airbyte CDK CLI."""
+
+from dataclasses import dataclass
+
+
+@dataclass(kw_only=True)
+class ConnectorSecretWithNoValidVersionsError(Exception):
+    """Error when a connector secret has no valid versions."""
+
+    connector_name: str
+    secret_name: str
+    gcp_project_id: str
+
+    def __str__(self) -> str:
+        """Return a string representation of the exception."""
+        from airbyte_cdk.cli.airbyte_cdk._secrets import _get_secret_url
+
+        url = _get_secret_url(self.secret_name, self.gcp_project_id)
+        return (
+            f"No valid versions found for secret '{self.secret_name}' in connector '{self.connector_name}'. "
+            f"Please check the following URL for more information:\n- {url}"
+        )

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1421,12 +1421,22 @@ definitions:
         default: ""
       schema_loader:
         title: Schema Loader
-        description: Component used to retrieve the schema for the current stream.
+        description:
+          One or many schema loaders can be used to retrieve the schema for the current stream. When
+          multiple schema loaders are defined, schema properties will be merged together. Schema
+          loaders defined first taking precedence in the event of a conflict.
         anyOf:
           - "$ref": "#/definitions/InlineSchemaLoader"
           - "$ref": "#/definitions/DynamicSchemaLoader"
           - "$ref": "#/definitions/JsonFileSchemaLoader"
           - "$ref": "#/definitions/CustomSchemaLoader"
+          - type: array
+            items:
+              anyOf:
+                - "$ref": "#/definitions/InlineSchemaLoader"
+                - "$ref": "#/definitions/DynamicSchemaLoader"
+                - "$ref": "#/definitions/JsonFileSchemaLoader"
+                - "$ref": "#/definitions/CustomSchemaLoader"
       # TODO we have move the transformation to the RecordSelector level in the code but kept this here for
       # compatibility reason. We should eventually move this to align with the code.
       transformations:
@@ -4362,4 +4372,4 @@ interpolation:
         regex: The regular expression to search for. It must include a capture group.
       return_type: str
       examples:
-        - '{{ "goodbye, cruel world" | regex_search("goodbye,\s(.*)$") }} -> "cruel world"'
+        - '{{ "goodbye, cruel world" | regex_search("goodbye,\s(.*)$") }} -> "cruel world"'

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1,3 +1,5 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
 # generated by datamodel-codegen:
 #   filename:  declarative_component_schema.yaml
 
@@ -2151,6 +2153,14 @@ class Config:
             DynamicSchemaLoader,
             JsonFileSchemaLoader,
             CustomSchemaLoader,
+            List[
+                Union[
+                    InlineSchemaLoader,
+                    DynamicSchemaLoader,
+                    JsonFileSchemaLoader,
+                    CustomSchemaLoader,
+                ]
+            ],
         ]
     ] = Field(
         None,

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -504,6 +504,7 @@
     SchemaTypeIdentifier,
     TypesMap,
 )
+from airbyte_cdk.sources.declarative.schema.composite_schema_loader import CompositeSchemaLoader
 from airbyte_cdk.sources.declarative.spec import Spec
 from airbyte_cdk.sources.declarative.stream_slicers import StreamSlicer
 from airbyte_cdk.sources.declarative.transformations import (
@@ -1914,9 +1915,25 @@ def create_declarative_stream(
         else:
             state_transformations = []
 
-        if model.schema_loader:
+        schema_loader: Union[
+            CompositeSchemaLoader,
+            DefaultSchemaLoader,
+            DynamicSchemaLoader,
+            InlineSchemaLoader,
+            JsonFileSchemaLoader,
+        ]
+        if model.schema_loader and isinstance(model.schema_loader, list):
+            nested_schema_loaders = [
+                self._create_component_from_model(model=nested_schema_loader, config=config)
+                for nested_schema_loader in model.schema_loader
+            ]
+            schema_loader = CompositeSchemaLoader(
+                schema_loaders=nested_schema_loaders, parameters={}
+            )
+        elif model.schema_loader:
             schema_loader = self._create_component_from_model(
-                model=model.schema_loader, config=config
+                model=model.schema_loader,  # type: ignore # If defined, schema_loader is guaranteed not to be a list and will be one of the existing base models
+                config=config,
             )
         else:
             options = model.parameters or {}

airbyte_cdk/sources/declarative/schema/composite_schema_loader.py

@@ -0,0 +1,31 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import InitVar, dataclass
+from typing import Any, Dict, List, Mapping
+
+from airbyte_cdk.sources.declarative.schema.schema_loader import SchemaLoader
+
+
+@dataclass
+class CompositeSchemaLoader(SchemaLoader):
+    """
+    Schema loader that consists of multiple schema loaders that are combined into a single
+    schema. Subsequent schemas do not overwrite existing values so the schema loaders with
+    a higher priority should be defined first.
+    """
+
+    schema_loaders: List[SchemaLoader]
+    parameters: InitVar[Mapping[str, Any]]
+
+    def get_json_schema(self) -> Mapping[str, Any]:
+        combined_schema: Dict[str, Any] = {
+            "$schema": "http://json-schema.org/draft-07/schema#",
+            "type": ["null", "object"],
+            "properties": {},
+        }
+        for schema_loader in self.schema_loaders:
+            schema_properties = schema_loader.get_json_schema()["properties"]
+            combined_schema["properties"] = {**schema_properties, **combined_schema["properties"]}
+        return combined_schema

airbyte_cdk/test/entrypoint_wrapper.py

@@ -82,6 +82,10 @@ def records(self) -> List[AirbyteMessage]:
     def state_messages(self) -> List[AirbyteMessage]:
         return self._get_message_by_types([Type.STATE])
 
+    @property
+    def spec_messages(self) -> List[AirbyteMessage]:
+        return self._get_message_by_types([Type.SPEC])
+
     @property
     def connection_status_messages(self) -> List[AirbyteMessage]:
         return self._get_message_by_types([Type.CONNECTION_STATUS])

airbyte_cdk/test/standard_tests/_job_runner.py

@@ -61,12 +61,15 @@ def spec(self, logger: logging.Logger) -> ConnectorSpecification:
 
 def run_test_job(
     connector: IConnector | type[IConnector] | Callable[[], IConnector],
-    verb: Literal["read", "check", "discover"],
-    test_scenario: ConnectorTestScenario,
+    verb: Literal["spec", "read", "check", "discover"],
     *,
+    test_scenario: ConnectorTestScenario | None = None,
     catalog: ConfiguredAirbyteCatalog | dict[str, Any] | None = None,
 ) -> entrypoint_wrapper.EntrypointOutput:
     """Run a test scenario from provided CLI args and return the result."""
+    # Use default (empty) scenario if not provided:
+    test_scenario = test_scenario or ConnectorTestScenario()
+
     if not connector:
         raise ValueError("Connector is required")
 
@@ -86,14 +89,14 @@ def run_test_job(
         )
 
     args: list[str] = [verb]
-    if test_scenario.config_path:
-        args += ["--config", str(test_scenario.config_path)]
-    elif test_scenario.config_dict:
+    config_dict = test_scenario.get_config_dict(empty_if_missing=True)
+    if config_dict and verb != "spec":
+        # Write the config to a temp json file and pass the path to the file as an argument.
         config_path = (
             Path(tempfile.gettempdir()) / "airbyte-test" / f"temp_config_{uuid.uuid4().hex}.json"
         )
         config_path.parent.mkdir(parents=True, exist_ok=True)
-        config_path.write_text(orjson.dumps(test_scenario.config_dict).decode())
+        config_path.write_text(orjson.dumps(config_dict).decode())
         args += ["--config", str(config_path)]
 
     catalog_path: Path | None = None