Merge branch 'main' into daryna/fix-substream-partition-router

Author: darynaishchenko

airbyte_cdk/cli/airbyte_cdk/_connector.py

@@ -45,8 +45,6 @@
 
 import rich_click as click
 
-from airbyte_cdk.test.standard_tests.util import create_connector_test_suite
-
 # from airbyte_cdk.test.standard_tests import pytest_hooks
 from airbyte_cdk.utils.connector_paths import (
     find_connector_root_from_name,
@@ -174,6 +172,9 @@ def run_connector_tests(
             "pytest is not installed. Please install pytest to run the connector tests."
         )
 
+    # Lazy import: this module requires pytest, which is only available via the [dev] extra.
+    from airbyte_cdk.test.standard_tests.util import create_connector_test_suite
+
     connector_test_suite = create_connector_test_suite(
         connector_name=connector_name if not connector_directory else None,
         connector_directory=connector_directory,

airbyte_cdk/cli/airbyte_cdk/_secrets.py

@@ -61,10 +61,12 @@
 logger = logging.getLogger("airbyte-cdk.cli.secrets")
 
 try:
+    import google.auth.exceptions
     from google.cloud import secretmanager_v1 as secretmanager
     from google.cloud.secretmanager_v1 import Secret
 except ImportError:
     # If the package is not installed, we will raise an error in the CLI command.
+    google = None  # type: ignore
     secretmanager = None  # type: ignore
     Secret = None  # type: ignore
 
@@ -235,7 +237,10 @@ def list_(
     """
     click.echo("Scanning secrets...", err=True)
 
-    connector_name, _ = resolve_connector_name_and_directory(connector)
+    if connector and isinstance(connector, str) and "/" not in connector and "\\" not in connector:
+        connector_name = connector
+    else:
+        connector_name, _ = resolve_connector_name_and_directory(connector)
     secrets: list[Secret] = _fetch_secret_handles(  # type: ignore
         connector_name=connector_name,
         gcp_project_id=gcp_project_id,
@@ -411,7 +416,14 @@ def _get_secret_filepath(
 
 
 def _get_gsm_secrets_client() -> "secretmanager.SecretManagerServiceClient":  # type: ignore
-    """Get the Google Secret Manager client."""
+    """Get the Google Secret Manager client.
+
+    If the `GCP_GSM_CREDENTIALS` environment variable is set, the client will be
+    created using service account credentials from that JSON string. Otherwise, the
+    client will fall back to Application Default Credentials (ADC), which supports
+    user credentials from `gcloud auth application-default login`, GCE metadata
+    server credentials, and other standard GCP authentication methods.
+    """
     if not secretmanager:
         raise ImportError(
             "google-cloud-secret-manager package is required for Secret Manager integration. "
@@ -420,18 +432,28 @@ def _get_gsm_secrets_client() -> "secretmanager.SecretManagerServiceClient":  #
         )
 
     credentials_json = os.environ.get("GCP_GSM_CREDENTIALS")
-    if not credentials_json:
-        raise ValueError(
-            "No Google Cloud credentials found. "
-            "Please set the `GCP_GSM_CREDENTIALS` environment variable."
+    if credentials_json:
+        click.echo(
+            "Using GCP service account credentials from GCP_GSM_CREDENTIALS env var.", err=True
+        )
+        return cast(
+            "secretmanager.SecretManagerServiceClient",
+            secretmanager.SecretManagerServiceClient.from_service_account_info(
+                json.loads(credentials_json)
+            ),
         )
 
-    return cast(
-        "secretmanager.SecretManagerServiceClient",
-        secretmanager.SecretManagerServiceClient.from_service_account_info(
-            json.loads(credentials_json)
-        ),
+    click.echo(
+        "GCP_GSM_CREDENTIALS not set. Using Application Default Credentials (ADC).", err=True
     )
+    try:
+        return secretmanager.SecretManagerServiceClient()
+    except google.auth.exceptions.DefaultCredentialsError:
+        raise ValueError(
+            "No Google Cloud credentials found. "
+            "Either set the `GCP_GSM_CREDENTIALS` environment variable with service account JSON, "
+            "or run `gcloud auth application-default login` to authenticate with your user account."
+        ) from None
 
 
 def _print_ci_secrets_masks(

airbyte_cdk/sources/declarative/interpolation/filters.py

@@ -136,6 +136,13 @@ def regex_search(value: str, regex: str) -> str:
     return ""
 
 
+def regex_replace(value: str, regex: str, replacement: str) -> str:
+    """
+    Replace all occurrences of a regular expression pattern in a string.
+    """
+    return re.sub(regex, replacement, value)
+
+
 def hmac(value: Any, key: str, hash_type: str = "sha256") -> str:
     """
     Implementation of a custom Jinja2 hmac filter with SHA-256 support.
@@ -181,6 +188,7 @@ def hmac(value: Any, key: str, hash_type: str = "sha256") -> str:
     base64binascii_decode,
     string,
     regex_search,
+    regex_replace,
     hmac,
 ]
 filters = {f.__name__: f for f in _filters_list}

airbyte_cdk/sources/declarative/resolvers/config_components_resolver.py

@@ -204,8 +204,10 @@ def _parse_yaml_if_possible(value: Any) -> Any:
                 return yaml.safe_load(value)
             except ParserError:  # "{{ record[0] in ['cohortActiveUsers'] }}"   # not valid YAML
                 return value
-            except ScannerError as e:  # "%Y-%m-%d'   # not valid yaml
+            except ScannerError as e:  # "%Y-%m-%d" or strings with tabs - not valid YAML
                 if "expected alphabetic or numeric character, but found '%'" in str(e):
                     return value
+                if "found character '\\t' that cannot start any token" in str(e):
+                    return value
                 raise e
         return value

airbyte_cdk/sources/file_based/stream/concurrent/adapters.py

@@ -155,10 +155,7 @@ def get_json_schema(self) -> Mapping[str, Any]:
 
     @property
     def primary_key(self) -> PrimaryKeyType:
-        return (
-            self._legacy_stream.config.primary_key
-            or self.get_parser().get_parser_defined_primary_key(self._legacy_stream.config)
-        )
+        return self._legacy_stream.primary_key
 
     def get_parser(self) -> FileTypeParser:
         return self._legacy_stream.get_parser()

airbyte_cdk/sources/file_based/stream/permissions_file_based_stream.py

@@ -7,6 +7,7 @@
 
 from airbyte_cdk.models import AirbyteLogMessage, AirbyteMessage, Level
 from airbyte_cdk.models import Type as MessageType
+from airbyte_cdk.sources.file_based.config.file_based_stream_config import PrimaryKeyType
 from airbyte_cdk.sources.file_based.file_based_stream_permissions_reader import (
     AbstractFileBasedStreamPermissionsReader,
 )
@@ -35,6 +36,10 @@ def __init__(
         super().__init__(**kwargs)
         self.stream_permissions_reader = stream_permissions_reader
 
+    @property
+    def primary_key(self) -> PrimaryKeyType:
+        return self.config.primary_key
+
     def _filter_schema_invalid_properties(
         self, configured_catalog_json_schema: Dict[str, Any]
     ) -> Dict[str, Any]:

unit_tests/sources/declarative/interpolation/test_filters.py

@@ -108,6 +108,36 @@ def test_regex_search_no_match() -> None:
     assert val is None
 
 
+@pytest.mark.parametrize(
+    "expression, expected",
+    [
+        pytest.param(
+            "{{ 'hello world' | regex_replace('world', 'there') }}",
+            "hello there",
+            id="basic_replacement",
+        ),
+        pytest.param(
+            "{{ 'abc123def456' | regex_replace('[0-9]+', '') }}",
+            "abcdef",
+            id="regex_pattern_strip_digits",
+        ),
+        pytest.param(
+            "{{ 'hello world' | regex_replace('xyz', 'replaced') }}",
+            "hello world",
+            id="no_match_returns_original",
+        ),
+        pytest.param(
+            "{{ 'aaa bbb aaa' | regex_replace('aaa', 'ccc') }}",
+            "ccc bbb ccc",
+            id="multiple_occurrences",
+        ),
+    ],
+)
+def test_regex_replace(expression: str, expected: str) -> None:
+    val = interpolation.eval(expression, {})
+    assert val == expected
+
+
 def test_hmac_sha256_default() -> None:
     message = "test_message"
     secret_key = "test_secret_key"

unit_tests/sources/declarative/resolvers/test_config_components_resolver.py

@@ -160,6 +160,20 @@ def to_configured_catalog(
     }
 )
 
+# Manifest with component definition with value containing tab characters
+# which would cause YAML ScannerError (tabs cannot start tokens in YAML)
+_MANIFEST_WITH_TAB_SCANNER_ERROR = deepcopy(_MANIFEST)
+_MANIFEST_WITH_TAB_SCANNER_ERROR["dynamic_streams"][0]["components_resolver"][
+    "components_mapping"
+].append(
+    {
+        "type": "ComponentMappingDefinition",
+        "create_or_update": True,
+        "field_path": ["retriever", "requester", "$parameters", "custom_query"],
+        "value": "SELECT\n\tcampaign.name,\n\tcampaign.id\nFROM campaign",  # Contains tab characters
+    }
+)
+
 
 @pytest.mark.parametrize(
     "manifest, config, expected_exception, expected_stream_names",
@@ -173,8 +187,15 @@ def to_configured_catalog(
         ),
         (_MANIFEST_WITH_STREAM_CONFIGS_LIST, _CONFIG, None, ["item_1", "item_2", "default_item"]),
         (_MANIFEST_WITH_SCANNER_ERROR, _CONFIG, None, ["item_1", "item_2", "default_item"]),
+        (_MANIFEST_WITH_TAB_SCANNER_ERROR, _CONFIG, None, ["item_1", "item_2", "default_item"]),
+    ],
+    ids=[
+        "no_duplicates",
+        "duplicates",
+        "stream_configs_list",
+        "scanner_error",
+        "tab_scanner_error",
     ],
-    ids=["no_duplicates", "duplicates", "stream_configs_list", "scanner_error"],
 )
 def test_dynamic_streams_read_with_config_components_resolver(
     manifest, config, expected_exception, expected_stream_names