osm_nbi/fslocal.py

   1 import os
   2 import logging
   3 import tarfile
   4 from http import HTTPStatus
   5 from shutil import rmtree
   6 from fsbase import FsBase, FsException
   7
   8 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
   9
  10
  11 class FsLocal(FsBase):
  12
  13     def __init__(self, logger_name='fs'):
  14         self.logger = logging.getLogger(logger_name)
  15         self.path = None
  16
  17     def get_params(self):
  18         return {"fs": "local", "path": self.path}
  19
  20     def fs_connect(self, config):
  21         try:
  22             if "logger_name" in config:
  23                 self.logger = logging.getLogger(config["logger_name"])
  24             self.path = config["path"]
  25             if not self.path.endswith("/"):
  26                 self.path += "/"
  27             if not os.path.exists(self.path):
  28                 raise FsException("Invalid configuration param at '[storage]': path '{}' does not exist".format(
  29                     config["path"]))
  30         except FsException:
  31             raise
  32         except Exception as e:  # TODO refine
  33             raise FsException(str(e))
  34
  35     def fs_disconnect(self):
  36         pass  # TODO
  37
  38     def mkdir(self, folder):
  39         """
  40         Creates a folder or parent object location
  41         :param folder:
  42         :return: None or raises and exception
  43         """
  44         try:
  45             os.mkdir(self.path + folder)
  46         except Exception as e:
  47             raise FsException(str(e), http_code=HTTPStatus.INTERNAL_SERVER_ERROR)
  48
  49     def file_exists(self, storage, mode=None):
  50         """
  51         Indicates if "storage" file exist
  52         :param storage: can be a str or a str list
  53         :param mode: can be 'file' exist as a regular file; 'dir' exists as a directory or; 'None' just exists
  54         :return: True, False
  55         """
  56         if isinstance(storage, str):
  57             f = storage
  58         else:
  59             f = "/".join(storage)
  60         if os.path.exists(self.path + f):
  61             if mode == "file" and os.path.isfile(self.path + f):
  62                 return True
  63             if mode == "dir" and os.path.isdir(self.path + f):
  64                 return True
  65         return False
  66
  67     def file_size(self, storage):
  68         """
  69         return file size
  70         :param storage: can be a str or a str list
  71         :return: file size
  72         """
  73         if isinstance(storage, str):
  74             f = storage
  75         else:
  76             f = "/".join(storage)
  77         return os.path.getsize(self.path + f)
  78
  79     def file_extract(self, tar_object, path):
  80         """
  81         extract a tar file
  82         :param tar_object: object of type tar
  83         :param path: can be a str or a str list, or a tar object where to extract the tar_object
  84         :return: None
  85         """
  86         if isinstance(path, str):
  87             f = self.path + path
  88         else:
  89             f = self.path + "/".join(path)
  90         tar_object.extractall(path=f)
  91
  92     def file_open(self, storage, mode):
  93         """
  94         Open a file
  95         :param storage: can be a str or list of str
  96         :param mode: file mode
  97         :return: file object
  98         """
  99         try:
 100             if isinstance(storage, str):
 101                 f = storage
 102             else:
 103                 f = "/".join(storage)
 104             return open(self.path + f, mode)
 105         except FileNotFoundError:
 106             raise FsException("File {} does not exist".format(f), http_code=HTTPStatus.NOT_FOUND)
 107         except IOError:
 108             raise FsException("File {} cannot be opened".format(f), http_code=HTTPStatus.BAD_REQUEST)
 109
 110     def dir_ls(self, storage):
 111         """
 112         return folder content
 113         :param storage: can be a str or list of str
 114         :return: folder content
 115         """
 116         try:
 117             if isinstance(storage, str):
 118                 f = storage
 119             else:
 120                 f = "/".join(storage)
 121             return os.listdir(self.path + f)
 122         except NotADirectoryError:
 123             raise FsException("File {} does not exist".format(f), http_code=HTTPStatus.NOT_FOUND)
 124         except IOError:
 125             raise FsException("File {} cannot be opened".format(f), http_code=HTTPStatus.BAD_REQUEST)
 126
 127     def file_delete(self, storage, ignore_non_exist=False):
 128         """
 129         Delete storage content recursivelly
 130         :param storage: can be a str or list of str
 131         :param ignore_non_exist: not raise exception if storage does not exist
 132         :return: None
 133         """
 134
 135         if isinstance(storage, str):
 136             f = self.path + storage
 137         else:
 138             f = self.path + "/".join(storage)
 139         if os.path.exists(f):
 140             rmtree(f)
 141         elif not ignore_non_exist:
 142             raise FsException("File {} does not exist".format(storage), http_code=HTTPStatus.NOT_FOUND)