added exists method for gcstorage

shub_workflow/deliver/futils.py

@@ -340,6 +340,8 @@ def exists(path, aws_key=None, aws_secret=None, aws_token=None, **kwargs):
     if check_s3_path(path):
         fs = S3FileSystem(**s3_credentials(aws_key, aws_secret, aws_token, region), **kwargs)
         return fs.exists(path)
+    if check_gcs_path(path):
+        return gcstorage.exists(path)
     return os_exists(path)
 
 

shub_workflow/deliver/gcstorage.py

@@ -141,3 +141,13 @@ def cp_file(src: str, dest: str):
     dest_bucket = storage_client.bucket(dest_bucket_name)
     bucket.copy_blob(bucket.blob(src_blob_name), dest_bucket, dest_blob_name)
     _LOGGER.info(f"Copied {src} to {dest}")
+
+
+def exists(src: str) -> bool:
+    storage_client = storage.Client()
+    m = _GS_FOLDER_RE.match(src)
+    assert m is not None, "Source must be in the format gs://<bucket>/<blob>"
+    src_bucket_name, src_blob_name = m.groups()
+
+    bucket = storage_client.bucket(src_bucket_name)
+    return bucket.blob(src_blob_name).exists()