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):
  50         """
  51         Indicates if "storage" file exist
  52         :param storage: can be a str or a str list
  53         :return: True, False
  54         """
  55         if isinstance(storage, str):
  56             f = storage
  57         else:
  58             f = "/".join(storage)
  59         return os.path.exists(self.path + f)
  60
  61     def file_size(self, storage):
  62         """
  63         return file size
  64         :param storage: can be a str or a str list
  65         :return: file size
  66         """
  67         if isinstance(storage, str):
  68             f = storage
  69         else:
  70             f = "/".join(storage)
  71         return os.path.getsize(self.path + f)
  72
  73     def file_extract(self, tar_object, path):
  74         """
  75         extract a tar file
  76         :param tar_object: object of type tar
  77         :param path: can be a str or a str list, or a tar object where to extract the tar_object
  78         :return: None
  79         """
  80         if isinstance(path, str):
  81             f = self.path + path
  82         else:
  83             f = self.path + "/".join(path)
  84         tar_object.extractall(path=f)
  85
  86     def file_open(self, storage, mode):
  87         """
  88         Open a file
  89         :param storage: can be a str or list of str
  90         :param mode: file mode
  91         :return: file object
  92         """
  93         if isinstance(storage, str):
  94             f = storage
  95         else:
  96             f = "/".join(storage)
  97         return open(self.path + f, mode)
  98
  99     def file_delete(self, storage, ignore_non_exist=False):
 100         """
 101         Delete storage content recursivelly
 102         :param storage: can be a str or list of str
 103         :param ignore_non_exist: not raise exception if storage does not exist
 104         :return: None
 105         """
 106
 107         if isinstance(storage, str):
 108             f = self.path + storage
 109         else:
 110             f = self.path + "/".join(storage)
 111         if os.path.exists(f):
 112             rmtree(f)
 113         elif not ignore_non_exist:
 114             raise FsException("File {} does not exist".format(storage), http_code=HTTPStatus.BAD_REQUEST)