def upload_single(self): blob_service_client = BlobServiceClient.from_connection_string(self.connection_string) download_links = {} for root, dirs, files in os.walk(self.folder): for file in files: full_path = os.path.join(root, file) # ignore hidden files if file.startswith("."): continue # if list_files is given, only upload matched files if self.list_files and file not in self.list_files: continue # if extension is given only upload if extension is matched if self.extension and os.path.isfile(full_path) and not file.lower().endswith(self.extension.lower()): continue blob_folder = root.replace(self.folder, "").lstrip("/") if self.blob_folder: # we only want to append blob_folder if it actually is a path or folder # blob_folder can be empty string "" if blob_folder: blob_folder = os.path.join(self.blob_folder, blob_folder) else: blob_folder = self.blob_folder # if no folder is given, just upload to the container root path if not blob_folder: container = self.destination else: container = os.path.join(self.destination, blob_folder) container_client = blob_service_client.get_container_client(container=container) with open(full_path, "rb") as data: logging.debug(f"Uploading blob {full_path}") container_client.upload_blob(data=data, name=file, overwrite=self.overwrite) if self.create_download_links: download_links[file] = self.create_blob_link(blob_folder=blob_folder, blob_name=file) return download_links