diff options
Diffstat (limited to 'azure/aria/aria-extension-cloudify/src/aria/aria/utils/archive.py')
-rw-r--r-- | azure/aria/aria-extension-cloudify/src/aria/aria/utils/archive.py | 66 |
1 files changed, 66 insertions, 0 deletions
diff --git a/azure/aria/aria-extension-cloudify/src/aria/aria/utils/archive.py b/azure/aria/aria-extension-cloudify/src/aria/aria/utils/archive.py new file mode 100644 index 0000000..29efcb1 --- /dev/null +++ b/azure/aria/aria-extension-cloudify/src/aria/aria/utils/archive.py @@ -0,0 +1,66 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Archive utilities. +""" + +import os +import tarfile +import zipfile +import tempfile +from contextlib import closing + + +def is_archive(source): + return tarfile.is_tarfile(source) or zipfile.is_zipfile(source) + + +def extract_archive(source): + if tarfile.is_tarfile(source): + return untar(source) + elif zipfile.is_zipfile(source): + return unzip(source) + raise ValueError( + 'Unsupported archive type provided or archive is not valid: {0}.'.format(source)) + + +def tar(source, destination): + with closing(tarfile.open(destination, 'w:gz')) as tar_archive: + tar_archive.add(source, arcname=os.path.basename(source)) + + +def untar(archive, destination=None): + if not destination: + destination = tempfile.mkdtemp() + with closing(tarfile.open(name=archive)) as tar_archive: + tar_archive.extractall(path=destination, members=tar_archive.getmembers()) + return destination + + +def zip(source, destination): + with closing(zipfile.ZipFile(destination, 'w')) as zip_file: + for root, _, files in os.walk(source): + for filename in files: + file_path = os.path.join(root, filename) + source_dir = os.path.dirname(source) + zip_file.write( + file_path, os.path.relpath(file_path, source_dir)) + return destination + + +def unzip(archive, destination=None): + if not destination: + destination = tempfile.mkdtemp() + with closing(zipfile.ZipFile(archive, 'r')) as zip_file: + zip_file.extractall(destination) + return destination |