Initialize the instance
(self, location=None, config_file="", log_level=None)
| 49 | return logging.getLogger(__name__) |
| 50 | |
| 51 | def __init__(self, location=None, config_file="", log_level=None) -> None: |
| 52 | """ |
| 53 | Initialize the instance |
| 54 | """ |
| 55 | |
| 56 | self._logger = self._log() |
| 57 | if log_level is not None: |
| 58 | self._logger.setLevel(log_level) |
| 59 | self._log_level = log_level |
| 60 | |
| 61 | if location not in [self.LOCAL_FILESYSTEM, self.AWS_S3, self.AZURE_BLOB]: |
| 62 | self._logger("Unknown logging location provided. Exiting") |
| 63 | exit(1) |
| 64 | |
| 65 | if config_file != "": |
| 66 | self._storage_config_file = config_file |
| 67 | |
| 68 | config = configparser.ConfigParser() |
| 69 | list = config.read(self._storage_config_file) |
| 70 | if len(list) == 0: |
| 71 | self._logger.error("Error: Could not find the config file") |
| 72 | exit(1) |
| 73 | |
| 74 | if location is not None and location != "": |
| 75 | self.storage_location = location |
| 76 | else: |
| 77 | self.storage_location = ConnectorUtil.get_config_setting( |
| 78 | self._logger, config, "DefaultStorage", "storage.location", "str", "" |
| 79 | ) |
| 80 | if self.storage_location is None or self.storage_location == "": |
| 81 | self.storage_location = self.LOCAL_FILESYSTEM |
| 82 | |
| 83 | # There is a classier (pun intended) way to do this. |
| 84 | # However, this is functional for now. |
| 85 | if self.storage_location == self.AZURE_BLOB: |
| 86 | self._instance = AzureStorageManager.AzureStorageManager( |
| 87 | config_file, log_level |
| 88 | ) |
| 89 | elif self.storage_location == self.AWS_S3: |
| 90 | self._instance = AWSStorageManager.AWSStorageManager(config_file, log_level) |
| 91 | else: |
| 92 | self._instance = LocalStorageManager.LocalStorageManager( |
| 93 | config_file, log_level |
| 94 | ) |
| 95 | |
| 96 | # Supported across all three classes |
| 97 | self.write_file = self._instance.write_file |
| 98 | self.read_file = self._instance.read_file |
| 99 | self.create_folder = self._instance.create_folder |
| 100 | self.write_large_file = self._instance.write_large_file |
| 101 | self.delete_file = self._instance.delete_file |
| 102 | self.list_directory = self._instance.list_directory |
nothing calls this directly
no test coverage detected