Merge branch 'main' into pnilan/feat/extend-spec-class-for-config-migrations

Author: pnilan

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/incremental/concurrent_partition_cursor.py

@@ -242,6 +242,10 @@ def _emit_state_message(self, throttle: bool = True) -> None:
             if current_time is None:
                 return
             self._last_emission_time = current_time
+            # Skip state emit for global cursor if parent state is empty
+            if self._use_global_cursor and not self._parent_state:
+                return
+
         self._connector_state_manager.update_state_for_stream(
             self._stream_name,
             self._stream_namespace,

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -27,6 +27,7 @@
 
 from isodate import parse_duration
 from pydantic.v1 import BaseModel
+from requests import Response
 
 from airbyte_cdk.connector_builder.models import (
     LogMessage as ConnectorBuilderLogMessage,
@@ -562,6 +563,7 @@
     PredicateValidator,
     ValidateAdheresToSchema,
 )
+from airbyte_cdk.sources.http_logger import format_http_message
 from airbyte_cdk.sources.message import (
     InMemoryMessageRepository,
     LogAppenderMessageRepositoryDecorator,
@@ -1582,7 +1584,6 @@ def create_concurrent_cursor_from_perpartition_cursor(
             )
         )
         stream_state = self.apply_stream_state_migrations(stream_state_migrations, stream_state)
-
         # Per-partition state doesn't make sense for GroupingPartitionRouter, so force the global state
         use_global_cursor = isinstance(
             partition_router, GroupingPartitionRouter
@@ -2490,15 +2491,24 @@ def create_dynamic_schema_loader(
                 schema_transformations.append(
                     self._create_component_from_model(model=transformation_model, config=config)
                 )
-
+        name = "dynamic_properties"
         retriever = self._create_component_from_model(
             model=model.retriever,
             config=config,
-            name="dynamic_properties",
+            name=name,
             primary_key=None,
             stream_slicer=combined_slicers,
             transformations=[],
             use_cache=True,
+            log_formatter=(
+                lambda response: format_http_message(
+                    response,
+                    f"Schema loader '{name}' request",
+                    f"Request performed in order to extract schema.",
+                    name,
+                    is_auxiliary=True,
+                )
+            ),
         )
         schema_type_identifier = self._create_component_from_model(
             model.schema_type_identifier, config=config, parameters=model.parameters or {}
@@ -3085,6 +3095,7 @@ def create_simple_retriever(
             ]
         ] = None,
         use_cache: Optional[bool] = None,
+        log_formatter: Optional[Callable[[Response], Any]] = None,
         **kwargs: Any,
     ) -> SimpleRetriever:
         def _get_url() -> str:
@@ -3261,6 +3272,7 @@ def _get_url() -> str:
                 config=config,
                 maximum_number_of_slices=self._limit_slices_fetched or 5,
                 ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
+                log_formatter=log_formatter,
                 parameters=model.parameters or {},
             )
         return SimpleRetriever(

airbyte_cdk/sources/declarative/requesters/http_job_repository.py

@@ -320,14 +320,14 @@ def _get_polling_response_interpolation_context(self, job: AsyncJob) -> Dict[str
         return polling_response_context
 
     def _get_create_job_stream_slice(self, job: AsyncJob) -> StreamSlice:
-        stream_slice = StreamSlice(
-            partition={},
-            cursor_slice={},
-            extra_fields={
+        return StreamSlice(
+            partition=job.job_parameters().partition,
+            cursor_slice=job.job_parameters().cursor_slice,
+            extra_fields=dict(job.job_parameters().extra_fields)
+            | {
                 "creation_response": self._get_creation_response_interpolation_context(job),
             },
         )
-        return stream_slice
 
     def _get_download_targets(self, job: AsyncJob) -> Iterable[str]:
         if not self.download_target_requester:

airbyte_cdk/sources/declarative/requesters/request_options/interpolated_nested_request_input_provider.py

@@ -11,6 +11,7 @@
 )
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.types import Config, StreamSlice
+from airbyte_cdk.utils.mapping_helpers import get_interpolation_context
 
 
 @dataclass
@@ -52,8 +53,8 @@ def eval_request_inputs(
         :param next_page_token: The pagination token
         :return: The request inputs to set on an outgoing HTTP request
         """
-        kwargs = {
-            "stream_slice": stream_slice,
-            "next_page_token": next_page_token,
-        }
+        kwargs = get_interpolation_context(
+            stream_slice=stream_slice,
+            next_page_token=next_page_token,
+        )
         return self._interpolator.eval(self.config, **kwargs)  # type: ignore  # self._interpolator is always initialized with a value and will not be None

airbyte_cdk/sources/declarative/requesters/request_options/interpolated_request_input_provider.py

@@ -8,6 +8,7 @@
 from airbyte_cdk.sources.declarative.interpolation.interpolated_mapping import InterpolatedMapping
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
+from airbyte_cdk.utils.mapping_helpers import get_interpolation_context
 
 
 @dataclass
@@ -51,10 +52,10 @@ def eval_request_inputs(
         :param valid_value_types: A tuple of types that the interpolator should allow
         :return: The request inputs to set on an outgoing HTTP request
         """
-        kwargs = {
-            "stream_slice": stream_slice,
-            "next_page_token": next_page_token,
-        }
+        kwargs = get_interpolation_context(
+            stream_slice=stream_slice,
+            next_page_token=next_page_token,
+        )
         interpolated_value = self._interpolator.eval(  # type: ignore # self._interpolator is always initialized with a value and will not be None
             self.config,
             valid_key_types=valid_key_types,