Tests/store_packs.py
def upload_to_storage(self, zip_pack_path, latest_version, storage_bucket, override_pack):
        """ Manages the upload of pack zip artifact to correct path in cloud storage.
        The zip pack will be uploaded to following path: /content/packs/pack_name/pack_latest_version.
        In case that zip pack artifact already exist at constructed path, the upload will be skipped.
        If flag override_pack is set to True, pack will forced for upload.

        Args:
            zip_pack_path (str): full path to pack zip artifact.
            latest_version (str): pack latest version.
            storage_bucket (google.cloud.storage.bucket.Bucket): google cloud storage bucket.
            override_pack (bool): whether to override existing pack.

        Returns:
            bool: whether the operation succeeded.
            bool: True in case of pack existence at targeted path and upload was skipped, otherwise returned False.

        """
        task_status = True

        try:
            version_pack_path = os.path.join(STORAGE_BASE_PATH, self._pack_name, latest_version)
            existing_files = [f.name for f in storage_bucket.list_blobs(prefix=version_pack_path)]

            if existing_files and not override_pack:
                print_warning(f"The following packs already exist at storage: {', '.join(existing_files)}")
                print_warning(f"Skipping step of uploading {self._pack_name}.zip to storage.")
                return task_status, True

            pack_full_path = f"{version_pack_path}/{self._pack_name}.zip"
            blob = storage_bucket.blob(pack_full_path)

            with open(zip_pack_path, "rb") as pack_zip:
                blob.upload_from_file(pack_zip)

            self.relative_storage_path = blob.name
            print_color(f"Uploaded {self._pack_name} pack to {pack_full_path} path.", LOG_COLORS.GREEN)

            return task_status, False
        except Exception as e:
            task_status = False
            print_error(f"Failed in uploading {self._pack_name} pack to gcs.\nAdditional info: {e}")
            return task_status, True
 Manages the upload of pack zip artifact to correct path in cloud storage.
        The zip pack will be uploaded to following path: /content/packs/pack_name/pack_latest_version.
        In case that zip pack artifact already exist at constructed path, the upload will be skipped.
        If flag override_pack is set to True, pack will forced for upload.

        Args:
            zip_pack_path (str): full path to pack zip artifact.
            latest_version (str): pack latest version.
            storage_bucket (google.cloud.storage.bucket.Bucket): google cloud storage bucket.
            override_pack (bool): whether to override existing pack.

        Returns:
            bool: whether the operation succeeded.
            bool: True in case of pack existence at targeted path and upload was skipped, otherwise returned False.

        
Manages the upload of pack zip artifact to correct path in cloud storage.
The zip pack will be uploaded to following path: /content/packs/pack_name/pack_latest_version.
In case that zip pack artifact already exist at constructed path, the upload will be skipped.
If flag override_pack is set to True, pack will forced for upload.
Manages the upload of pack zip artifact to correct path in cloud storage . The zip pack will be uploaded to following path : / content / packs / pack_name / pack_latest_version . In case that zip pack artifact already exist at constructed path the upload will be skipped . If flag override_pack is set to True pack will forced for upload .
def upload_to_storage(self, zip_pack_path, latest_version, storage_bucket, override_pack):
        task_status = True
        try:
            version_pack_path = os.path.join(STORAGE_BASE_PATH, self._pack_name, latest_version)
            existing_files = [f.name for f in storage_bucket.list_blobs(prefix=version_pack_path)]
            if existing_files and not override_pack:
                print_warning(f"The following packs already exist at storage: {', '.join(existing_files)}")
                print_warning(f"Skipping step of uploading {self._pack_name}.zip to storage.")
                return task_status, True
            pack_full_path = f"{version_pack_path}/{self._pack_name}.zip"
            blob = storage_bucket.blob(pack_full_path)
            with open(zip_pack_path, "rb") as pack_zip:
                blob.upload_from_file(pack_zip)
            self.relative_storage_path = blob.name
            print_color(f"Uploaded {self._pack_name} pack to {pack_full_path} path.", LOG_COLORS.GREEN)
            return task_status, False
        except Exception as e:
            task_status = False
            print_error(f"Failed in uploading {self._pack_name} pack to gcs.\nAdditional info: {e}")
            return task_status, True
def upload_to_storage ( self , zip_pack_path , latest_version , storage_bucket , override_pack ) : task_status = True try : version_pack_path = os . path . join ( STORAGE_BASE_PATH , self . _pack_name , latest_version ) existing_files = [ f . name for f in storage_bucket . list_blobs ( prefix = version_pack_path ) ] if existing_files and not override_pack : print_warning ( f"The following packs already exist at storage: {', '.join(existing_files)}" ) print_warning ( f"Skipping step of uploading {self._pack_name}.zip to storage." ) return task_status , True pack_full_path = f"{version_pack_path}/{self._pack_name}.zip" blob = storage_bucket . blob ( pack_full_path ) with open ( zip_pack_path , "rb" ) as pack_zip : blob . upload_from_file ( pack_zip ) self . relative_storage_path = blob . name print_color ( f"Uploaded {self._pack_name} pack to {pack_full_path} path." , LOG_COLORS . GREEN ) return task_status , False except Exception as e : task_status = False print_error ( f"Failed in uploading {self._pack_name} pack to gcs.\nAdditional info: {e}" ) return task_status , True
Manages the upload of pack zip artifact to correct path in cloud storage.
Manages the upload of pack zip artifact to correct path in cloud storage .
""" Manages the upload of pack zip artifact to correct path in cloud storage.
        The zip pack will be uploaded to following path: /content/packs/pack_name/pack_latest_version.
        In case that zip pack artifact already exist at constructed path, the upload will be skipped.
        If flag override_pack is set to True, pack will forced for upload.

        Args:
            zip_pack_path (str): full path to pack zip artifact.
            latest_version (str): pack latest version.
            storage_bucket (google.cloud.storage.bucket.Bucket): google cloud storage bucket.
            override_pack (bool): whether to override existing pack.

        Returns:
            bool: whether the operation succeeded.
            bool: True in case of pack existence at targeted path and upload was skipped, otherwise returned False.

        """
whether the operation succeeded.
bool: True in case of pack existence at targeted path and upload was skipped, otherwise returned False.
whether the operation succeeded . bool : True in case of pack existence at targeted path and upload was skipped otherwise returned False .
bool
self
zip_pack_path
full path to pack zip artifact.
full path to pack zip artifact .
latest_version
pack latest version.
pack latest version .
storage_bucket
google cloud storage bucket.
google cloud storage bucket .
override_pack
whether to override existing pack.
whether to override existing pack .