Cloud#

Helpful utilities for working with cloud services.

Convenience functions related to cloud infrastructure.

class cpg_utils.cloud.DockerImage(name, uri, tag_uri, size, build_time)[source]#

cpg_utils.cloud.email_from_id_token(id_token_jwt: str) → str[source]#

Decodes the ID token (JWT) to get the email address of the caller.

See for details: https://developers.google.com/identity/sign-in/web/backend-auth?authuser=0#verify-the-integrity-of-the-id-token

This function assumes that the token has been verified beforehand.

cpg_utils.cloud.find_image(repository: str | None, name: str, version: str) → DockerImage[source]#: Returns image details or raises ValueError if the image or tag does not exist.

cpg_utils.cloud.get_google_identity_token(target_audience: str | None, request: Request | None = None) → str[source]#: Returns a Google identity token for the given audience.

cpg_utils.cloud.get_path_components_from_gcp_path(path: str) → dict[str, str][source]#: Return the {bucket_name}, {dataset}, {bucket_type}, {subdir}, and {file} for GS only paths Uses regex to match the full bucket name, dataset name, bucket type (e.g. ‘test’, ‘main-upload’, ‘release’), subdirectory, and the file name.

cpg_utils.cloud.get_project_id_from_service_account_email(service_account_email: str) → str[source]#

Get GCP project id from service_account_email

>>> get_project_id_from_service_account_email('cromwell-test@tob-wgs.iam.gserviceaccount.com')
'tob-wgs'

cpg_utils.cloud.read_secret(project_id: str, secret_name: str, fail_gracefully: bool = True) → str | None[source]#

Reads the latest version of a GCP Secret Manager secret.

Returns None if the secret doesn’t exist or there was a problem retrieving it, unless fail_gracefully is set to False.

cpg_utils.cloud.write_secret(project_id: str, secret_name: str, secret_value: str) → None[source]#

Adds a new version for a GCP Secret Manager secret and disables all previous versions

Parameters:

Cloud