def copy()

in fbpcp/service/storage_gcs.py [0:0]
59 lines of code
15 McCabe index (conditional complexity)

    def copy(self, source: str, destination: str, recursive: bool = False) -> None:
        """Move a file or folder between local storage and GCS, or between GCS and GCS

        Args:
            source: source file
            destination: destination file
            recursive: whether to recursively copy a folder

        Raises:
            ValueError: Source is folder and recursive is False
            ValueError: Source is not a valid Local or GCS path
            ValueError: Destination is not a valid Local or GCS path
            ValueError: Both source and destination are the same
            ValueError: Both source and destination are local
        """
        sourceType = StorageService.path_type(source)
        destinationType = StorageService.path_type(destination)
        if sourceType != PathType.GCS and sourceType != PathType.Local:
            raise ValueError("Source is not a valid Local or GCS path")
        if destinationType != PathType.GCS and destinationType != PathType.Local:
            raise ValueError("Destination is not a valid Local or GCS path")
        if source == destination:
            raise ValueError("Both source and destination are the same")

        # source is Local
        if sourceType == PathType.Local:
            if destinationType == PathType.Local:
                raise ValueError("Both source and destination are local files")
            else:
                gcs_path = GCSPath(destination)
                if recursive:
                    self.upload_dir(
                        source=source,
                        gcs_path_bucket=gcs_path.bucket,
                        gcs_path_key=gcs_path.key,
                    )
                else:
                    if os.path.isdir(source):
                        raise ValueError(
                            f"Source {source} is a folder. Use --recursive"
                        )
                    self.gcs_gateway.upload_file(
                        file_name=source, bucket=gcs_path.bucket, key=gcs_path.key
                    )
        # source is GCS
        elif sourceType == PathType.GCS:
            source_gcs_path = GCSPath(source)
            if destinationType == PathType.Local:
                if recursive:
                    self.download_dir(
                        gcs_path_bucket=source_gcs_path.bucket,
                        gcs_path_key=source_gcs_path.key,
                        destination=destination,
                    )
                else:
                    self.gcs_gateway.download_file(
                        bucket=source_gcs_path.bucket,
                        key=source_gcs_path.key,
                        filename=destination,
                    )
            elif destinationType == PathType.GCS:
                destination_gcs_path = GCSPath(destination)
                if recursive:
                    self.copy_dir(
                        source_bucket=source_gcs_path.bucket,
                        source_key=source_gcs_path.key,
                        destination_bucket=destination_gcs_path.bucket,
                        destination_key=destination_gcs_path.key,
                    )
                else:
                    self.gcs_gateway.copy(
                        source_bucket_name=source_gcs_path.bucket,
                        source_key=source_gcs_path.key,
                        dest_bucket_name=destination_gcs_path.bucket,
                        dest_key=destination_gcs_path.key,
                    )