Skip to content

Commit

Permalink
Feat: Add singular fetch_connector_secret() for `GoogleGSMSecretMan…
Browse files Browse the repository at this point in the history
…ager`, fix GSM edge cases, and add CI tests (#191)
  • Loading branch information
aaronsteers authored Apr 12, 2024
1 parent 77497c5 commit 97119c9
Show file tree
Hide file tree
Showing 3 changed files with 105 additions and 8 deletions.
57 changes: 49 additions & 8 deletions airbyte/secrets/google_gsm.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,10 +130,16 @@ def __init__(

def get_secret(self, secret_name: str) -> SecretString | None:
"""Get a named secret from Google Colab user secrets."""
full_name = secret_name
if "projects/" not in full_name:
# This is not yet fully qualified
full_name = f"projects/{self.project}/secrets/{secret_name}/versions/latest"

if "/versions/" not in full_name:
full_name += "/versions/latest"

return SecretString(
self.secret_client.access_secret_version(
name=f"projects/{self.project}/secrets/{secret_name}/versions/latest"
).payload.data.decode("UTF-8")
self.secret_client.access_secret_version(name=full_name).payload.data.decode("UTF-8")
)

def fetch_secrets(
Expand All @@ -155,11 +161,10 @@ def fetch_secrets(
Iterable[SecretHandle]: An iterable of `SecretHandle` objects for the matching secrets.
"""
gsm_secrets: ListSecretsPager = self.secret_client.list_secrets(
secretmanager.ListSecretsRequest(
request={
"filter": filter_string,
}
)
request=secretmanager.ListSecretsRequest(
filter=filter_string,
parent=f"projects/{self.project}",
),
)

return [
Expand Down Expand Up @@ -205,3 +210,39 @@ def fetch_connector_secrets(
label_key=self.CONNECTOR_LABEL,
label_value=connector_name,
)

def fetch_connector_secret(
self,
connector_name: str,
) -> SecretHandle:
"""Fetch secret in the secret manager, using the connector name as a filter for the label.
This method is a convenience method that returns the first secret found for the connector.
The label key used to filter the secrets is defined by the `CONNECTOR_LABEL` attribute,
which defaults to 'connector'.
Args:
connector_name (str): The name of the connector to filter by.
Returns:
SecretHandle: The matching secret.
"""
results: Iterable[SecretHandle] = self.fetch_connector_secrets(connector_name)
try:
result = next(iter(results))
except StopIteration:
raise exc.PyAirbyteError(
message="No secrets found for connector.",
guidance=(
"Please check that the connector name is correct "
"and that the secret is correctly labeled."
),
context={
"project": self.project,
"connector_name": connector_name,
"label_key": self.CONNECTOR_LABEL,
},
) from None

return result
Empty file.
56 changes: 56 additions & 0 deletions tests/integration_tests/secrets/test_gsm_secrets.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
"""Tests for the GSM secrets manager."""
from __future__ import annotations

from airbyte.secrets.base import SecretHandle
from airbyte.secrets.google_gsm import GoogleGSMSecretManager


def test_get_gsm_secret(ci_secret_manager: GoogleGSMSecretManager) -> dict:
assert ci_secret_manager.get_secret(
"SECRET_DESTINATION_DUCKDB__MOTHERDUCK__CREDS",
).parse_json()


def test_get_gsm_secrets_with_filter(ci_secret_manager: GoogleGSMSecretManager) -> None:
"""Test fetching connector secrets."""
secrets = ci_secret_manager.fetch_secrets(
filter_string="labels.connector=source-bigquery",
)
assert secrets is not None
secrets_list = list(secrets)
assert len(secrets_list) > 0
assert secrets_list[0].get_value().is_json()


def test_get_gsm_secrets_by_label(ci_secret_manager: GoogleGSMSecretManager) -> None:
"""Test fetching connector secrets."""
secrets = ci_secret_manager.fetch_secrets_by_label(
label_key="connector",
label_value="source-salesforce",
)
assert secrets is not None
secrets_list = list(secrets)
assert len(secrets_list) > 0
assert secrets_list[0].get_value().is_json()


def test_get_connector_secrets(ci_secret_manager: GoogleGSMSecretManager) -> None:
"""Test fetching connector secrets."""
secrets = ci_secret_manager.fetch_connector_secrets(
"source-salesforce"
)
assert secrets is not None
secrets_list = list(secrets)
assert len(secrets_list) > 0
assert secrets_list[0].get_value().is_json()


def test_first_connector_secret(ci_secret_manager: GoogleGSMSecretManager) -> None:
"""Test fetching connector secrets."""
secret = ci_secret_manager.fetch_connector_secret(
"source-salesforce"
)
assert secret is not None
assert isinstance(secret, SecretHandle)
assert secret.get_value().is_json()

0 comments on commit 97119c9

Please sign in to comment.