SecretManagerUtils¶

SecretManagerUtils is a lightweight wrapper around the GCP Secret Manager API. It makes it easy to store and retrieve secrets (API keys, credentials, connection strings) from a notebook or automation script without manually constructing Secret Manager resource names.

A common pattern is to store a service account JSON as a Secret Manager secret, then retrieve it at runtime to authenticate other CXAS Scrapi classes — keeping secrets out of your source code and environment variables.

Quick Example¶

from cxas_scrapi import SecretManagerUtils

sm = SecretManagerUtils(project_id="my-gcp-project")

# Create a new secret (or get it if it already exists)
secret_version = sm.create_or_get_secret(
    secret_id="my-api-key",
    payload="super-secret-value-123",
)
print("Secret version:", secret_version)

# Retrieve the secret value later
value = sm.get_secret(secret_id="my-api-key")
print("Retrieved:", value)

# Use a secret to authenticate another class
import json
creds_json = sm.get_secret(secret_id="my-service-account-json")
creds_dict = json.loads(creds_json)

from cxas_scrapi import Sessions
sessions = Sessions(
    app_name="projects/my-project/locations/us/apps/my-app",
    creds_dict=creds_dict,
)

Reference¶

SecretManagerUtils ¶

SecretManagerUtils(project_id)

Utility Class for Creating and Retrieving Secret Manager Secrets.

Source code in src/cxas_scrapi/utils/secret_manager_utils.py

def __init__(self, project_id: str):
    self.project_id = project_id
    self.client = secretmanager.SecretManagerServiceClient()

create_or_get_secret ¶

create_or_get_secret(secret_id, payload=None)

Retrieves a Secret via Secret ID, or creates a new one if it doesn't exist.

Parameters:

Name	Type	Description	Default
`secret_id`	`str`	The ID of the Secret to retrieve or create.	required
`payload`	`str`	The string payload to add as a new Secret Version if creating a new secret.	`None`

Returns:

Type	Description
`str`	The full resource name path to the latest version of the
`str`	Secret, e.g.
`str`	`projects/{project_id}/secrets/{secret_id}/versions/latest`.

Source code in src/cxas_scrapi/utils/secret_manager_utils.py

def create_or_get_secret(self, secret_id: str, payload: str = None) -> str:
    """Retrieves a Secret via Secret ID, or creates a new one if it
    doesn't exist.

    Args:
        secret_id: The ID of the Secret to retrieve or create.
        payload: The string payload to add as a new Secret Version if
                 creating a new secret.

    Returns:
        The full resource name path to the latest version of the
        Secret, e.g.
        `projects/{project_id}/secrets/{secret_id}/versions/latest`.
    """
    parent = f"projects/{self.project_id}"

    # Check if secret already exists
    request = {"parent": parent}
    secrets = self.client.list_secrets(request=request)

    for secret in secrets:
        display_name = secret.name.split("/")[-1]
        if display_name == secret_id:
            print(f"Found existing secret: {secret_id}")
            return f"{secret.name}/versions/latest"

    # Secret doesn't exist, create it
    if payload is None:
        raise ValueError(
            "Secret does not exist and no payload was provided to "
            "create one."
        )

    print(f"Creating new secret: {secret_id}")
    created_secret = self.client.create_secret(
        request={
            "parent": parent,
            "secret_id": secret_id,
            "secret": {"replication": {"automatic": {}}},
        }
    )

    # Add the initial version (payload)
    payload_bytes = payload.encode("UTF-8")
    self.client.add_secret_version(
        request={
            "parent": created_secret.name,
            "payload": {"data": payload_bytes},
        }
    )

    return f"{created_secret.name}/versions/latest"

create_secret_with_version ¶

create_secret_with_version(secret_id, secret_payload)

Creates a secret if it doesn't exist, adds a new version with the provided payload, and returns the resource name of the new version.

Source code in src/cxas_scrapi/utils/secret_manager_utils.py

def create_secret_with_version(
    self, secret_id: str, secret_payload: str
) -> Optional[str]:
    """Creates a secret if it doesn't exist, adds a new version with the
    provided payload, and returns the resource name of the new version.
    """
    parent = f"projects/{self.project_id}"
    # Sanitize secret_id to meet GCP requirements
    safe_secret_id = re.sub(r"[^a-zA-Z0-9_-]", "_", secret_id)
    secret_name = f"{parent}/secrets/{safe_secret_id}"

    try:
        self.client.get_secret(request={"name": secret_name})
        logger.info(
            f"Secret '{safe_secret_id}' already exists. "
            "Adding a new version."
        )
    except api_exceptions.NotFound:
        logger.info(
            f"Secret '{safe_secret_id}' not found. Creating it now..."
        )
        try:
            self.client.create_secret(
                request={
                    "parent": parent,
                    "secret_id": safe_secret_id,
                    "secret": {"replication": {"automatic": {}}},
                }
            )
        except api_exceptions.AlreadyExists:
            logger.info(
                f"Secret '{safe_secret_id}' was created by another "
                "process. Continuing."
            )
        except Exception as e:
            logger.error(f"Failed to create secret '{safe_secret_id}': {e}")
            return None

    try:
        payload_bytes = secret_payload.encode("UTF-8")
        add_version_response = self.client.add_secret_version(
            request={
                "parent": secret_name,
                "payload": {"data": payload_bytes},
            }
        )
        logger.info(
            "-> Success! Created new secret version: "
            f"{add_version_response.name.split('/')[-1]}"
        )
        return add_version_response.name
    except Exception as e:
        logger.error(
            f"Failed to add version to secret '{safe_secret_id}': {e}"
        )
        return None