add metadata output to copy_bucket_to_gcs method

parsons/google/google_cloud_storage.py

@@ -1,21 +1,22 @@
 import datetime
 import gzip
 import logging
 import time
 import uuid
 import zipfile
 from typing import Optional, Union
-
+from collections import OrderedDict
 import google
 import petl
 from google.cloud import storage, storage_transfer
 from google.oauth2.credentials import Credentials
+from google.protobuf import json_format
 
 from parsons.google.utilities import (
     load_google_application_credentials,
     setup_google_application_credentials,
 )
 from parsons.utilities import files
 
 logger = logging.getLogger(__name__)
 
@@ -406,7 +407,7 @@
         source_path: str = "",
         aws_access_key_id: Optional[str] = None,
         aws_secret_access_key: Optional[str] = None,
-    ):
+    ) -> OrderedDict:
         """
         Creates a one-time transfer job from Amazon S3 to Google Cloud
         Storage. Copies all blobs within the bucket unless a key or prefix
@@ -426,6 +427,10 @@
                 Access key to authenticate storage transfer
             aws_secret_access_key (str):
                 Secret key to authenticate storage transfer
+        `Returns:`
+            OrderedDict:
+                An ordered dict containing the metadata of
+                the finished transfer operation
         """
         if source not in ["gcs", "s3"]:
             raise ValueError(f"Blob transfer only supports gcs and s3 sources [source={source}]")
@@ -510,18 +515,16 @@
                     logger.debug("Operation still running...")
 
                 else:
-                    operation_metadata = storage_transfer.TransferOperation.deserialize(
-                        operation.metadata.value
-                    )
-                    error_output = operation_metadata.error_breakdowns
-                    if len(error_output) != 0:
+                    operation_metadata = json_format.MessageToDict(operation.metadata)
+                    error_output = operation_metadata.get("errorBreakdowns", None)
+                    if error_output:
                         raise Exception(
                             f"""{blob_storage} to GCS Transfer Job
                             {create_result.name} failed with error: {error_output}"""
                         )
                     else:
                         logger.info(f"TransferJob: {create_result.name} succeeded.")
-                        return
+                        return operation_metadata
 
             else:
                 logger.info("Waiting to kickoff operation...")