cidc_api/shared/gcloud_client.py
"""Utilities for interacting with the Google Cloud Platform APIs."""
import json
import os
from cidc_api.config.secrets import get_secrets_manager
os.environ["TZ"] = "UTC"
import datetime
import warnings
import hashlib
from collections import namedtuple
from concurrent.futures import Future
from sqlalchemy.orm.session import Session
from typing import (
Any,
BinaryIO,
Callable,
Dict,
Iterable,
List,
Optional,
Set,
Tuple,
Union,
)
import requests
from google.cloud import storage, pubsub, bigquery
from google.cloud.bigquery.enums import EntityTypes
from google.oauth2.service_account import Credentials
from werkzeug.datastructures import FileStorage
import googleapiclient.discovery
from google.api_core.iam import Policy
from ..config.settings import (
GOOGLE_INTAKE_ROLE,
GOOGLE_INTAKE_BUCKET,
GOOGLE_UPLOAD_ROLE,
GOOGLE_UPLOAD_BUCKET,
GOOGLE_UPLOAD_TOPIC,
GOOGLE_ACL_DATA_BUCKET,
GOOGLE_LISTER_ROLE,
GOOGLE_BIGQUERY_USER_ROLE,
GOOGLE_CLOUD_PROJECT,
GOOGLE_EMAILS_TOPIC,
GOOGLE_PATIENT_SAMPLE_TOPIC,
GOOGLE_ARTIFACT_UPLOAD_TOPIC,
GOOGLE_GRANT_DOWNLOAD_PERMISSIONS_TOPIC,
TESTING,
ENV,
DEV_CFUNCTIONS_SERVER,
INACTIVE_USER_DAYS,
)
from ..config.logging import get_logger
from cidc_schemas.prism.constants import ASSAY_TO_FILEPATH
logger = get_logger(__name__)
_storage_client = None
_bigquery_client = None
_crm_service = None
def _get_storage_client() -> storage.Client:
"""
the project which the client acts on behalf of falls back to the default inferred from the environment
see: https://googleapis.dev/python/storage/latest/client.html#google.cloud.storage.client.Client
directly providing service account credentials for signing in get_signed_url() below
"""
global _storage_client
if _storage_client is None:
secret_manager = get_secrets_manager()
credentials = Credentials.from_service_account_info(
json.loads(secret_manager.get("APP_ENGINE_CREDENTIALS"))
)
_storage_client = storage.Client(credentials=credentials)
return _storage_client
def _get_crm_service() -> googleapiclient.discovery.Resource:
"""
Initializes a Cloud Resource Manager service.
"""
global _crm_service
if _crm_service is None:
secret_manager = get_secrets_manager()
credentials = Credentials.from_service_account_info(
json.loads(secret_manager.get("APP_ENGINE_CREDENTIALS"))
)
_crm_service = googleapiclient.discovery.build(
"cloudresourcemanager", "v1", credentials=credentials
)
return _crm_service
def _get_bucket(bucket_name: str) -> storage.Bucket:
"""
Get the bucket with name `bucket_name` from GCS.
This does not make an HTTP request; it simply instantiates a bucket object owned by _storage_client.
see: https://googleapis.dev/python/storage/latest/client.html#google.cloud.storage.client.Client.bucket
"""
storage_client = _get_storage_client()
bucket = storage_client.bucket(bucket_name)
return bucket
def _get_project_policy() -> Policy:
"""
Get the project policy.
"""
crm_service = _get_crm_service()
policy = (
crm_service.projects()
.getIamPolicy(
resource=GOOGLE_CLOUD_PROJECT,
body={},
)
.execute()
)
return policy
def _get_bigquery_dataset(dataset_id: str) -> bigquery.Dataset:
"""
Get the bigquery dataset with the id 'dataset_id'.
makes an API request to pull this with the bigquery client
"""
global _bigquery_client
if _bigquery_client is None:
secret_manager = get_secrets_manager()
credentials = Credentials.from_service_account_info(
json.loads(secret_manager.get("APP_ENGINE_CREDENTIALS"))
)
_bigquery_client = bigquery.Client(credentials=credentials)
dataset = _bigquery_client.get_dataset(dataset_id) # Make an API request.
return dataset
_xlsx_gcs_uri_format = (
"{trial_id}/xlsx/{template_category}/{template_type}/{upload_moment}.xlsx"
)
_pseudo_blob = namedtuple(
"_pseudo_blob", ["name", "size", "md5_hash", "crc32c", "time_created"]
)
def upload_xlsx_to_gcs(
trial_id: str,
template_category: str,
template_type: str,
filebytes: BinaryIO,
upload_moment: str,
) -> storage.Blob:
"""
Upload an xlsx template file to GOOGLE_ACL_DATA_BUCKET, returning the object URI.
`template_category` is either "manifests" or "assays".
`template_type` is an assay or manifest type, like "wes" or "pbmc" respectively.
Returns:
arg1: GCS blob object
"""
blob_name = _xlsx_gcs_uri_format.format(
trial_id=trial_id,
template_category=template_category,
template_type=template_type,
upload_moment=upload_moment,
)
if ENV == "dev":
logger.info(
f"Would've saved {blob_name} to {GOOGLE_UPLOAD_BUCKET} and {GOOGLE_ACL_DATA_BUCKET}"
)
return _pseudo_blob(
blob_name, 0, "_pseudo_md5_hash", "_pseudo_crc32c", upload_moment
)
upload_bucket: storage.Bucket = _get_bucket(GOOGLE_UPLOAD_BUCKET)
blob = upload_bucket.blob(blob_name)
filebytes.seek(0)
blob.upload_from_file(filebytes)
data_bucket = _get_bucket(GOOGLE_ACL_DATA_BUCKET)
final_object = upload_bucket.copy_blob(blob, data_bucket)
return final_object
def grant_lister_access(user_email: str) -> None:
"""
Grant a user list access to the GOOGLE_ACL_DATA_BUCKET. List access is
required for the user to download or read objects from this bucket.
As lister is an IAM permission on an ACL-controlled bucket, can't have conditions.
"""
logger.info(f"granting list to {user_email}")
bucket = _get_bucket(GOOGLE_ACL_DATA_BUCKET)
grant_storage_iam_access(bucket, GOOGLE_LISTER_ROLE, user_email, expiring=False)
def revoke_lister_access(user_email: str) -> None:
"""
Revoke a user's list access to the GOOGLE_ACL_DATA_BUCKET. List access is
required for the user to download or read objects from this bucket.
Unlike grant_lister_access, revoking doesn't care if the binding is expiring or not so we don't need to specify.
"""
logger.info(f"revoking list to {user_email}")
bucket = _get_bucket(GOOGLE_ACL_DATA_BUCKET)
revoke_storage_iam_access(bucket, GOOGLE_LISTER_ROLE, user_email)
def grant_upload_access(user_email: str) -> None:
"""
Grant a user upload access to the GOOGLE_UPLOAD_BUCKET. Upload access
means a user can write objects to the bucket but cannot delete,
overwrite, or read objects from this bucket.
Non-expiring as GOOGLE_UPLOAD_BUCKET is subject to ACL.
"""
logger.info(f"granting upload to {user_email}")
bucket = _get_bucket(GOOGLE_UPLOAD_BUCKET)
grant_storage_iam_access(bucket, GOOGLE_UPLOAD_ROLE, user_email, expiring=False)
def revoke_upload_access(user_email: str) -> None:
"""
Revoke a user's upload access from GOOGLE_UPLOAD_BUCKET.
"""
logger.info(f"revoking upload from {user_email}")
bucket = _get_bucket(GOOGLE_UPLOAD_BUCKET)
revoke_storage_iam_access(bucket, GOOGLE_UPLOAD_ROLE, user_email)
def grant_bigquery_access(user_emails: List[str]) -> None:
"""
Grant a user's access to run bigquery queries on project.
Grant access to public level bigquery tables.
"""
logger.info(f"granting bigquery access to {user_emails}")
policy = _get_project_policy()
grant_bigquery_iam_access(policy, user_emails)
def revoke_bigquery_access(user_email: str) -> None:
"""
Revoke a user's access to run bigquery queries on project.
Revoke access to public level bigquery tables.
"""
logger.info(f"revoking bigquery access from {user_email}")
policy = _get_project_policy()
revoke_bigquery_iam_access(policy, user_email)
def get_intake_bucket_name(user_email: str) -> str:
"""
Get the name for an intake bucket associated with the given user.
Bucket names will have a structure like GOOGLE_INTAKE_BUCKET-<hash>
"""
# 10 characters should be plenty, given that we only expect
# a handful of unique data uploaders - we get 16^10 possible hashes.
email_hash = hashlib.sha1(bytes(user_email, "utf-8")).hexdigest()[:10]
bucket_name = f"{GOOGLE_INTAKE_BUCKET}-{email_hash}"
return bucket_name
def create_intake_bucket(user_email: str) -> storage.Bucket:
"""
Create a new data intake bucket for this user, or get the existing one.
Grant the user GCS object admin permissions on the bucket, or refresh those
permissions if they've already been granted.
Created with uniform bucket-level IAM access, so expiring permission.
"""
storage_client = _get_storage_client()
bucket_name = get_intake_bucket_name(user_email)
bucket = storage_client.bucket(bucket_name)
# For Data Freeze only return bucket if it already existsG
if bucket.exists():
return bucket
else:
return None
# if not bucket.exists():
# # Create a new bucket with bucket-level permissions enabled.
# bucket.iam_configuration.uniform_bucket_level_access_enabled = True
# bucket = storage_client.create_bucket(bucket)
# # Grant the user appropriate permissions
# grant_storage_iam_access(bucket, GOOGLE_INTAKE_ROLE, user_email)
# return bucket
def refresh_intake_access(user_email: str) -> None:
"""
Re-grant a user's access to their intake bucket if it exists.
"""
bucket_name = get_intake_bucket_name(user_email)
bucket = _get_bucket(bucket_name)
if bucket.exists():
grant_storage_iam_access(bucket, GOOGLE_INTAKE_ROLE, user_email)
def revoke_intake_access(user_email: str) -> None:
"""
Re-grant a user's access to their intake bucket if it exists.
"""
bucket_name = get_intake_bucket_name(user_email)
bucket = _get_bucket(bucket_name)
if bucket.exists():
revoke_storage_iam_access(bucket, GOOGLE_INTAKE_ROLE, user_email)
def upload_xlsx_to_intake_bucket(
user_email: str, trial_id: str, upload_type: str, xlsx: FileStorage
) -> str:
"""
Upload a metadata spreadsheet file to the GCS intake bucket,
returning the URL to the bucket in the GCP console.
"""
# add a timestamp to the metadata file name to avoid overwriting previous versions
filename_with_ts = f'{xlsx.filename.rsplit(".xlsx", 1)[0]}_{datetime.datetime.now().isoformat()}.xlsx'
blob_name = f"{trial_id}/{upload_type}/metadata/{filename_with_ts}"
# upload the metadata spreadsheet to the intake bucket
bucket_name = get_intake_bucket_name(user_email)
bucket = _get_bucket(bucket_name)
blob = bucket.blob(blob_name)
blob.upload_from_file(xlsx)
return f"https://console.cloud.google.com/storage/browser/_details/{bucket_name}/{blob_name}"
def _execute_multiblob_acl_change(
user_email_list: List[str],
blob_list: List[storage.Blob],
callback_fn: Callable[[storage.acl._ACLEntity], None],
) -> None:
"""
Spools out each blob and each user with saving the blob.
callback_fn is called on each blob / user to make the changes in permissions there.
See see https://googleapis.dev/python/storage/latest/acl.html
After processing all of the users for each blob, blob.acl.save() is called.
Parameters
----------
user_email_list : List[str]
blob_list: List[google.cloud.storage.Blob]
used to generate blob / user ACL entries
callback_fun : Callable[google.cloud.storage.acl._ACLEntity]
each blob / user ACL entry is passed in turn
"""
for blob in blob_list:
for user_email in user_email_list:
blob_user = blob.acl.user(user_email)
callback_fn(blob_user)
blob.acl.save()
def get_blob_names(
trial_id: Optional[str],
upload_type: Optional[Tuple[str]],
session: Optional[Session] = None,
) -> Set[str]:
"""session only needed if trial_id is None"""
prefixes: Set[str] = _build_trial_upload_prefixes(
trial_id, upload_type, session=session
)
# https://googleapis.dev/python/storage/latest/client.html#google.cloud.storage.client.Client.list_blobs
blob_list = []
storage_client = _get_storage_client()
for prefix in prefixes:
blob_list.extend(
storage_client.list_blobs(GOOGLE_ACL_DATA_BUCKET, prefix=prefix)
)
return set([blob.name for blob in blob_list])
def grant_download_access_to_blob_names(
user_email_list: List[str],
blob_name_list: List[str],
) -> None:
"""
Using ACL, grant download access to all blobs given to the user(s) given.
"""
bucket = _get_bucket(GOOGLE_ACL_DATA_BUCKET)
blob_list = [bucket.get_blob(name) for name in blob_name_list]
if isinstance(user_email_list, str):
user_email_list = [user_email_list]
_execute_multiblob_acl_change(
user_email_list=user_email_list,
blob_list=blob_list,
callback_fn=lambda obj: obj.grant_read(),
)
def grant_download_access(
user_email_list: Union[List[str], str],
trial_id: Optional[str],
upload_type: Optional[Union[str, List[str]]],
) -> None:
"""
Gives users download access to all objects in a trial of a particular upload type.
If trial_id is None, then grant access to all trials.
If upload_type is None, then grant access to all upload_types.
if user_email_list is []. then CFn loads users from db table.
If the user already has download access for this trial and upload type, idempotent.
Download access is controlled by IAM on production and ACL elsewhere.
"""
user_email_list = (
[user_email_list] if isinstance(user_email_list, str) else user_email_list
)
logger.info(
f"Granting download access on trial {trial_id} upload {upload_type} to {user_email_list}"
)
# ---- Handle through main grant permissions topic ----
# would time out in CFn
kwargs = {
"trial_id": trial_id,
"upload_type": upload_type,
"user_email_list": user_email_list,
"revoke": False,
}
report = _encode_and_publish(str(kwargs), GOOGLE_GRANT_DOWNLOAD_PERMISSIONS_TOPIC)
# Wait for response from pub/sub
if report:
report.result()
def revoke_download_access_from_blob_names(
user_email_list: List[str],
blob_name_list: List[str],
) -> None:
"""
Using ACL, grant download access to all blobs given to the users given.
"""
bucket = _get_bucket(GOOGLE_ACL_DATA_BUCKET)
blob_list = [bucket.get_blob(name) for name in blob_name_list]
def revoke(blob_user: storage.acl._ACLEntity):
blob_user.revoke_owner()
blob_user.revoke_write()
blob_user.revoke_read()
_execute_multiblob_acl_change(
blob_list=blob_list,
callback_fn=revoke,
user_email_list=user_email_list,
)
def revoke_download_access(
user_email_list: Union[str, List[str]],
trial_id: Optional[str],
upload_type: Optional[Union[str, List[str]]],
) -> None:
"""
Revoke users' download access to all objects in a trial of a particular upload type.
If trial_id is None, then revoke access to all trials.
If upload_type is None, then revoke access to all upload_types.
if user_email_list is []. then CFn loads users from db table.
Return the GCS URIs from which access has been revoked.
Download access is controlled by ACL.
"""
user_email_list = (
[user_email_list] if isinstance(user_email_list, str) else user_email_list
)
logger.info(
f"Revoking download access on trial {trial_id} upload {upload_type} from {user_email_list}"
)
# ---- Handle through main grant permissions topic ----
# would timeout in cloud function
kwargs = {
"trial_id": trial_id,
"upload_type": upload_type,
"user_email_list": user_email_list,
"revoke": True,
}
report = _encode_and_publish(str(kwargs), GOOGLE_GRANT_DOWNLOAD_PERMISSIONS_TOPIC)
# Wait for response from pub/sub
if report:
report.result()
def _build_trial_upload_prefixes(
trial_id: Optional[str],
upload_type: Optional[Tuple[Optional[str]]],
session: Optional[Session] = None,
) -> Set[str]:
"""
Build the set of prefixes associated with the trial_id and upload_type
If no trial_id is given, all trials are used.
If no upload_type is given, the prefixes are everything but clinical_data.
If upload_type has no files, returns empty set.
if None in upload_type, it's treated the same as bare None
If neither are given, an empty set is returned.
session is only used with trial_id is None and upload_type is not None
"""
if trial_id is None and (upload_type is None or None in upload_type):
return set()
trial_set: Set[str] = set()
upload_set: Set[str] = set()
if not trial_id:
from ..models.models import TrialMetadata
trial_set = set(
[
str(t.trial_id)
for t in session.query(TrialMetadata).add_columns(
TrialMetadata.trial_id
)
]
)
else:
trial_set = set([trial_id])
if not upload_type or None in upload_type:
upload_set = {
upload_name
for upload_name in ASSAY_TO_FILEPATH.keys()
if upload_name != "clinical_data"
}
else:
upload_set = set(upload_type)
ret: Set[str] = set()
for trial in trial_set:
for upload in upload_set:
if upload_type:
if upload in ASSAY_TO_FILEPATH:
ret.add(f"{trial}/{ASSAY_TO_FILEPATH[upload]}")
else: # null means cross-assay
# don't affect clinical_data
ret = ret.union(
{
f"{trial}/{upload_prefix}"
for trial in trial_set
for upload_name, upload_prefix in ASSAY_TO_FILEPATH.items()
if upload_name != "clinical_data"
}
)
return ret
def grant_storage_iam_access(
bucket: storage.Bucket,
role: str,
user_email: str,
expiring: bool = True,
) -> None:
"""
Grant `user_email` the provided IAM `role` on a storage `bucket`.
Default assumes `bucket` is IAM controlled and should expire after `INACTIVE_USER_DAYS` days have elapsed.
Set `expiring` to False for IAM permissions on ACL-controlled buckets.
"""
# see https://cloud.google.com/storage/docs/access-control/using-iam-permissions#code-samples_3
policy = bucket.get_iam_policy(requested_policy_version=3)
policy.version = 3
# remove the existing binding if one exists so that we can recreate it with an updated TTL.
_find_and_pop_storage_iam_binding(policy, role, user_email)
if not expiring:
# special value -1 for non-expiring
binding = _build_storage_iam_binding(bucket.name, role, user_email, ttl_days=-1)
else:
binding = _build_storage_iam_binding(
bucket.name, role, user_email
) # use default
# insert the binding into the policy
policy.bindings.append(binding)
try:
bucket.set_iam_policy(policy)
except Exception as e:
logger.error(str(e))
raise e
def grant_bigquery_iam_access(policy: Policy, user_emails: List[str]) -> None:
"""
Grant all 'user_emails' the "roles/bigquery.jobUser" role on project.
If we are in the production environment, all 'user_emails' also get access to
the public bigquery dataset in prod.
"""
roles = [b["role"] for b in policy["bindings"]]
if (
GOOGLE_BIGQUERY_USER_ROLE in roles
): # if the role is already in the policy, add the users
binding = next(
b for b in policy["bindings"] if b["role"] == GOOGLE_BIGQUERY_USER_ROLE
)
for user_email in user_emails:
binding["members"].append(user_member(user_email))
else: # otherwise create the role and add to policy
binding = {
"role": GOOGLE_BIGQUERY_USER_ROLE,
"members": [
user_member(user_email) for user_email in user_emails
], # convert format
}
policy["bindings"].append(binding)
# try to set the new policy with edits
try:
_crm_service.projects().setIamPolicy(
resource=GOOGLE_CLOUD_PROJECT,
body={
"policy": policy,
},
).execute()
except Exception as e:
logger.error(str(e))
raise e
# grant dataset level access to public dataset
dataset_id = GOOGLE_CLOUD_PROJECT + ".public"
dataset = _get_bigquery_dataset(dataset_id)
entries = list(dataset.access_entries)
for user_email in user_emails:
entries.append(
bigquery.AccessEntry(
role="READER",
entity_type=EntityTypes.USER_BY_EMAIL,
entity_id=user_email,
)
)
dataset.access_entries = entries
_bigquery_client.update_dataset(dataset, ["access_entries"]) # Make an API request.
# Arbitrary upper bound on the number of GCS IAM bindings we expect a user to have for uploads
MAX_REVOKE_ALL_ITERATIONS = 250
def revoke_storage_iam_access(
bucket: storage.Bucket, role: str, user_email: str
) -> None:
"""Revoke a bucket IAM policy made by calling `grant_storage_iam_access`."""
# see https://cloud.google.com/storage/docs/access-control/using-iam-permissions#code-samples_3
policy = bucket.get_iam_policy(requested_policy_version=3)
policy.version = 3
# find and remove any matching policy binding for this user
for i in range(MAX_REVOKE_ALL_ITERATIONS):
removed_binding = _find_and_pop_storage_iam_binding(policy, role, user_email)
if removed_binding is None:
if i == 0:
warnings.warn(
f"Tried to revoke a non-existent download IAM permission for {user_email}"
)
break
try:
bucket.set_iam_policy(policy)
except Exception as e:
logger.error(str(e))
raise e
def revoke_bigquery_iam_access(policy: Policy, user_email: str) -> None:
"""
Revoke 'user_email' the "roles/bigquery.jobUser" role on project.
If we are in the production environment, 'user_email' also get access
revoked from the public bigquery dataset in prod.
"""
# find and remove user on binding
binding = next(
b for b in policy["bindings"] if b["role"] == GOOGLE_BIGQUERY_USER_ROLE
)
if "members" in binding and user_member(user_email) in binding["members"]:
binding["members"].remove(user_member(user_email))
# try update of the policy
try:
policy = (
_crm_service.projects()
.setIamPolicy(
resource=GOOGLE_CLOUD_PROJECT,
body={
"policy": policy,
},
)
.execute()
)
except Exception as e:
logger.error(str(e))
raise e
# remove dataset level access
dataset_id = GOOGLE_CLOUD_PROJECT + ".public"
dataset = _get_bigquery_dataset(dataset_id)
entries = list(dataset.access_entries)
dataset.access_entries = [
entry for entry in entries if entry.entity_id != user_email
]
dataset = _bigquery_client.update_dataset(
dataset,
# Update just the `access_entries` property of the dataset.
["access_entries"],
) # Make an API request.
user_member = lambda email: f"user:{email}"
def _build_storage_iam_binding(
bucket: str,
role: str,
user_email: str,
ttl_days: int = INACTIVE_USER_DAYS,
) -> Dict[str, Any]:
"""
Grant the user associated with `user_email` the provided IAM `role` when acting
on objects in `bucket`. This permission remains active for `ttl_days` days.
See GCP common expression language syntax overview: https://cloud.google.com/iam/docs/conditions-overview
Parameters
----------
bucket: str
the name of the bucket to build the binding for
role: str
the role name to build the binding for
user_email: str
the email of the user to build the binding for
ttl_days: int = INACTIVE_USER_DAYS
the number of days until this permission should expire
pass -1 for non-expiring
Returns
-------
List[dict]
the bindings to be put onto policy.bindings
"""
timestamp = datetime.datetime.now()
expiry_date = (timestamp + datetime.timedelta(ttl_days)).date()
# going to add the expiration condition after, so don't return directly
ret = {
"role": role,
"members": {user_member(user_email)}, # convert format
}
if ttl_days >= 0:
# special value -1 doesn't expire
ret["condition"] = {
"title": f"{role} access on {bucket}",
"description": f"Auto-updated by the CIDC API on {timestamp}",
"expression": f'request.time < timestamp("{expiry_date.isoformat()}T00:00:00Z")',
}
return ret
def _find_and_pop_storage_iam_binding(
policy: storage.bucket.Policy,
role: str,
user_email: str,
) -> Optional[dict]:
"""
Find an IAM policy binding for the given `user_email`, `policy`, and `role`, and pop
it from the policy's bindings list if it exists.
"""
# try to find the policy binding on the `policy`
user_binding_index = None
for i, binding in enumerate(policy.bindings):
role_matches = binding.get("role") == role
member_matches = binding.get("members") == {user_member(user_email)}
if role_matches and member_matches:
# a user should be a member of no more than one conditional download binding
# if they do, warn - but use the last one because this isn't breaking
if user_binding_index is not None:
warnings.warn(
f"Found multiple conditional bindings for {user_email} role {role}. This is an invariant violation - "
"check out permissions on the CIDC GCS buckets to debug."
)
user_binding_index = i
binding = (
policy.bindings.pop(user_binding_index)
if user_binding_index is not None
else None
)
return binding
def get_signed_url(
object_name: str,
bucket_name: str = GOOGLE_ACL_DATA_BUCKET,
method: str = "GET",
expiry_mins: int = 30,
) -> str:
"""
Generate a signed URL for `object_name` to give a client temporary access.
Using v2 signed urls because v4 is in Beta and response_disposition doesn't work.
https://cloud.google.com/storage/docs/access-control/signing-urls-with-helpers
"""
storage_client = _get_storage_client()
bucket = storage_client.get_bucket(bucket_name)
blob = bucket.blob(object_name)
# Generate the signed URL, allowing a client to use `method` for `expiry_mins` minutes
expiration = datetime.timedelta(minutes=expiry_mins)
full_filename = object_name.replace("/", "_").replace('"', "_").replace(" ", "_")
url = blob.generate_signed_url(
version="v2",
expiration=expiration,
method=method,
response_disposition=f'attachment; filename="{full_filename}"',
)
logger.info(f"generated signed URL for {object_name}: {url}")
return url
def _encode_and_publish(content: str, topic: str) -> Future:
"""Convert `content` to bytes and publish it to `topic`."""
pubsub_publisher = pubsub.PublisherClient()
topic = pubsub_publisher.topic_path(GOOGLE_CLOUD_PROJECT, topic)
data = bytes(content, "utf-8")
# Don't actually publish to Pub/Sub if running locally
if ENV == "dev":
if DEV_CFUNCTIONS_SERVER:
logger.info(
f"Publishing message {content!r} to topic {DEV_CFUNCTIONS_SERVER}/{topic}"
)
import base64
bdata = base64.b64encode(content.encode("utf-8"))
try:
res = requests.post(
f"{DEV_CFUNCTIONS_SERVER}/{topic}", data={"data": bdata}
)
except Exception as e:
raise Exception(
f"Couldn't publish message {content!r} to topic {DEV_CFUNCTIONS_SERVER}/{topic}"
) from e
else:
logger.info(f"Got {res}")
if res.status_code != 200:
raise Exception(
f"Couldn't publish message {content!r} to {DEV_CFUNCTIONS_SERVER}/{topic}: {res!r}"
)
else:
logger.info(f"Would've published message {content} to topic {topic}")
return
# The Pub/Sub publisher client returns a concurrent.futures.Future
# containing info about whether the publishing was successful.
report = pubsub_publisher.publish(topic, data=data)
return report
def publish_upload_success(job_id: int) -> None:
"""Publish to the uploads topic that the upload job with the provided `job_id` succeeded."""
report = _encode_and_publish(str(job_id), GOOGLE_UPLOAD_TOPIC)
# For now, we wait await this Future. Going forward, maybe
# we should look for a way to leverage asynchrony here.
if report:
report.result()
def publish_patient_sample_update(manifest_upload_id: int) -> None:
"""Publish to the patient_sample_update topic that a new manifest has been uploaded."""
report = _encode_and_publish(str(manifest_upload_id), GOOGLE_PATIENT_SAMPLE_TOPIC)
# Wait for response from pub/sub
if report:
report.result()
def publish_artifact_upload(file_id: int) -> None:
"""Publish a downloadable file ID to the artifact_upload topic"""
report = _encode_and_publish(str(file_id), GOOGLE_ARTIFACT_UPLOAD_TOPIC)
# Wait for response from pub/sub
if report:
report.result()
def send_email(to_emails: List[str], subject: str, html_content: str, **kw) -> None:
"""
Publish an email-to-send to the emails topic.
`kw` are expected to be sendgrid json api style additional email parameters.
"""
# Don't actually send an email if this is a test
if TESTING or ENV == "dev":
logger.info(f"Would send email with subject '{subject}' to {to_emails}")
return
email_json = json.dumps(
dict(to_emails=to_emails, subject=subject, html_content=html_content, **kw)
)
report = _encode_and_publish(email_json, GOOGLE_EMAILS_TOPIC)
# Await confirmation that the published message was received.
if report:
report.result()