lcm/osm_common/fslocal.py

   1 import os
   2 import tarfile
   3 from http import HTTPStatus
   4 from shutil import rmtree
   5 from fsbase import FsBase, FsException
   6
   7
   8 class FsLocal(FsBase):
   9
  10     def __init__(self):
  11         self.path = None
  12
  13     def get_params(self):
  14         return {"fs": "local", "path": self.path}
  15
  16     def fs_connect(self, config):
  17         try:
  18             self.path = config["path"]
  19             if not self.path.endswith("/"):
  20                 self.path += "/"
  21             if not os.path.exists(self.path):
  22                 raise FsException("Invalid configuration param at '[storage]': path '{}' does not exist".format(
  23                     config["path"]))
  24         except FsException:
  25             raise
  26         except Exception as e:  # TODO refine
  27             raise FsException(str(e))
  28
  29     def fs_disconnect(self):
  30         pass  # TODO
  31
  32     def mkdir(self, folder):
  33         """
  34         Creates a folder or parent object location
  35         :param folder:
  36         :return: None or raises and exception
  37         """
  38         try:
  39             os.mkdir(self.path + folder)
  40         except Exception as e:
  41             raise FsException(str(e), http_code=HTTPStatus.INTERNAL_SERVER_ERROR.value)
  42
  43     def file_exists(self, storage):
  44         """
  45         Indicates if "storage" file exist
  46         :param storage: can be a str or a str list
  47         :return: True, False
  48         """
  49         if isinstance(storage, str):
  50             f = storage
  51         else:
  52             f = "/".join(storage)
  53         return os.path.exists(self.path + f)
  54
  55     def file_size(self, storage):
  56         """
  57         return file size
  58         :param storage: can be a str or a str list
  59         :return: file size
  60         """
  61         if isinstance(storage, str):
  62             f = storage
  63         else:
  64             f = "/".join(storage)
  65         return os.path.getsize(self.path + f)
  66
  67     def file_extract(self, tar_object, path):
  68         """
  69         extract a tar file
  70         :param tar_object: object of type tar
  71         :param path: can be a str or a str list, or a tar object where to extract the tar_object
  72         :return: None
  73         """
  74         if isinstance(path, str):
  75             f = self.path + path
  76         else:
  77             f = self.path + "/".join(path)
  78         tar_object.extractall(path=f)
  79
  80     def file_open(self, storage, mode):
  81         """
  82         Open a file
  83         :param storage: can be a str or list of str
  84         :param mode: file mode
  85         :return: file object
  86         """
  87         if isinstance(storage, str):
  88             f = storage
  89         else:
  90             f = "/".join(storage)
  91         return open(self.path + f, mode)
  92
  93     def file_delete(self, storage, ignore_non_exist=False):
  94         """
  95         Delete storage content recursivelly
  96         :param storage: can be a str or list of str
  97         :param ignore_non_exist: not raise exception if storage does not exist
  98         :return: None
  99         """
 100
 101         if isinstance(storage, str):
 102             f = self.path + storage
 103         else:
 104             f = self.path + "/".join(storage)
 105         if os.path.exists(f):
 106             rmtree(f)
 107         elif not ignore_non_exist:
 108             raise FsException("File {} does not exist".format(storage), http_code=HTTPStatus.BAD_REQUEST.value)