|
| 1 | + |
| 2 | +import os |
| 3 | +import re |
| 4 | +from azure.storage.file import FileService |
| 5 | +from config import Configuration |
| 6 | + |
| 7 | + |
| 8 | +class AzureFileManager(): |
| 9 | + def __init__(self): |
| 10 | + # fetch config data |
| 11 | + conf = Configuration() |
| 12 | + # create Azure File share service |
| 13 | + self.file_service = FileService( |
| 14 | + account_name=conf.account_name, account_key=conf.account_key) |
| 15 | + # set azure share file name (container) |
| 16 | + self.file_share = conf.file_share |
| 17 | + |
| 18 | + def upload_file(self, upload_path, file_path): |
| 19 | + if not os.path.isfile(file_path): |
| 20 | + print("Your file is not exists, check your file path and try again.") |
| 21 | + return |
| 22 | + filename = os.path.basename(file_path) |
| 23 | + # remove ' or " from path, if path was empty like "" or '' set upload_path=None, this make upload file to root directory |
| 24 | + upload_path = upload_path.strip().replace("'", '').replace('"', '') |
| 25 | + # remove start and end / or \ |
| 26 | + if upload_path.endswith('/') or upload_path.endswith('\\'): |
| 27 | + upload_path = upload_path[:-1] |
| 28 | + if upload_path.startswith('/') or upload_path.startswith('\\'): |
| 29 | + upload_path = upload_path[1:] |
| 30 | + # sanity check |
| 31 | + upload_path = upload_path if len(upload_path) >= 1 else None |
| 32 | + |
| 33 | + print("Start uploading...") |
| 34 | + try: |
| 35 | + # create sub directories |
| 36 | + self.create_sub_directories(upload_path) |
| 37 | + # upload |
| 38 | + self.file_service.create_file_from_path( |
| 39 | + share_name=self.file_share, # file_share name in azure |
| 40 | + directory_name=upload_path, # server directories address. None => root directory |
| 41 | + file_name=filename, # Name of file to create in azure |
| 42 | + local_file_path=file_path) |
| 43 | + print("'{0}' has been successfully uploaded".format(filename)) |
| 44 | + except: |
| 45 | + print("Failed to upload '{0}', please try again".format(filename)) |
| 46 | + |
| 47 | + def download_file(self, file_path): |
| 48 | + """ download file from azure, enter file path in azure """ |
| 49 | + # check file path was not empty |
| 50 | + file_path = file_path.strip().replace("'", '').replace('"', '') |
| 51 | + if len(file_path) == 0: |
| 52 | + print("Please enter a file path") |
| 53 | + return |
| 54 | + filename = os.path.basename(file_path) |
| 55 | + dir_path = os.path.dirname(file_path) |
| 56 | + # if parent path was not available, use None => root directory |
| 57 | + dir_path = dir_path if dir_path else None |
| 58 | + |
| 59 | + print("Downloading...") |
| 60 | + try: |
| 61 | + self.file_service.get_file_to_path( |
| 62 | + share_name=self.file_share, |
| 63 | + directory_name=dir_path, # The path to the directory in azure |
| 64 | + file_name=filename, # Name of existing file in azure |
| 65 | + # Path of file to write to local machine |
| 66 | + file_path="{0}".format(filename)) |
| 67 | + print( |
| 68 | + "'{0}' has been successfully downloaded and saved in current directory.".format(filename)) |
| 69 | + except: |
| 70 | + print("Failed to download '{0}', either file doesn't exist or you are offline.".format( |
| 71 | + filename)) |
| 72 | + |
| 73 | + def get_list_of_files(self, dir_name=None): |
| 74 | + """ show list of all files and all directories in azure""" |
| 75 | + generator = self.file_service.list_directories_and_files( |
| 76 | + share_name=self.file_share, |
| 77 | + directory_name=dir_name) |
| 78 | + parent = "" if dir_name == None else dir_name |
| 79 | + for file_or_dir in generator: |
| 80 | + if not re.match(r"(.[a-z]*[A-Z]*[0-9]*)$", file_or_dir.name): |
| 81 | + # file |
| 82 | + if len(parent) == 0: |
| 83 | + print(file_or_dir.name) |
| 84 | + else: |
| 85 | + print("{0}/{1}".format(parent, file_or_dir.name)) |
| 86 | + else: |
| 87 | + # dir |
| 88 | + if len(parent) == 0: |
| 89 | + self.get_list_of_files(file_or_dir.name) |
| 90 | + else: |
| 91 | + self.get_list_of_files( |
| 92 | + "{0}/{1}".format(parent, file_or_dir.name)) |
| 93 | + |
| 94 | + def create_sub_directories(self, path): |
| 95 | + """ create sub directories in Azure """ |
| 96 | + if path is None: |
| 97 | + return |
| 98 | + dirs = os.path.normpath(path).split(os.path.sep) |
| 99 | + parent = '' |
| 100 | + for dir in dirs: |
| 101 | + parent += dir if len(parent) == 0 else '/'+dir |
| 102 | + self.file_service.create_directory(self.file_share, parent) |
0 commit comments