def copy()

in fbpcp/service/storage_s3.py [0:0]

• 52 lines of code
• 11 McCabe index (conditional complexity)

    def copy(self, source: str, destination: str, recursive: bool = False) -> None:
        """Move a file or folder between local storage and S3, as well as, S3 and S3
        Keyword arguments:
        source -- source file
        destination -- destination file
        recursive -- whether to recursively copy a folder
        """
        if StorageService.path_type(source) == PathType.Local:
            # from local to S3
            if StorageService.path_type(destination) == PathType.Local:
                raise ValueError("Both source and destination are local files")
            s3_path = S3Path(destination)
            if path.isdir(source):
                if not recursive:
                    raise ValueError(f"Source {source} is a folder. Use --recursive")
                self.upload_dir(source, s3_path.bucket, s3_path.key)
            else:
                self.s3_gateway.upload_file(source, s3_path.bucket, s3_path.key)
        else:
            source_s3_path = S3Path(source)
            if StorageService.path_type(destination) == PathType.S3:
                # from S3 to S3
                dest_s3_path = S3Path(destination)
                if source_s3_path == dest_s3_path:
                    raise ValueError(
                        f"Source {source} and destination {destination} are the same"
                    )

                if source.endswith("/"):
                    if not recursive:
                        raise ValueError(
                            f"Source {source} is a folder. Use --recursive"
                        )

                    self.copy_dir(
                        source_s3_path.bucket,
                        source_s3_path.key + "/",
                        dest_s3_path.bucket,
                        dest_s3_path.key,
                    )
                else:
                    self.s3_gateway.copy(
                        source_s3_path.bucket,
                        source_s3_path.key,
                        dest_s3_path.bucket,
                        dest_s3_path.key,
                    )
            else:
                # from S3 to local
                if source.endswith("/"):
                    if not recursive:
                        raise ValueError(
                            f"Source {source} is a folder. Use --recursive"
                        )
                    self.download_dir(
                        source_s3_path.bucket,
                        source_s3_path.key + "/",
                        destination,
                    )
                else:
                    self.s3_gateway.download_file(
                        source_s3_path.bucket, source_s3_path.key, destination
                    )