Saimon/refactor filesystem api (#293)

Author: saimonation

cterasdk/asynchronous/core/files/browser.py

@@ -1,5 +1,5 @@
 from ....cio.core import CorePath
-from ....lib import FileSystem
+from ....lib.storage import asynfs, commonfs
 from ..base_command import BaseCommand
 from . import io
 
@@ -9,7 +9,6 @@ class FileBrowser(BaseCommand):
     def __init__(self, core):
         super().__init__(core)
         self._scope = f'/{self._core.context}/webdav'
-        self._filesystem = FileSystem.instance()
 
     async def handle(self, path):
         """
@@ -30,6 +29,38 @@ async def handle_many(self, directory, *objects):
         handle_many_function = await io.handle_many(self.normalize(directory), *objects)
         return await handle_many_function(self._core)
 
+    async def download(self, path, destination=None):
+        """
+        Download a file
+
+        :param str path: Path
+        :param str,optional destination:
+         File destination, if it is a directory, the original filename will be kept, defaults to the default directory
+        """
+        directory, name = commonfs.determine_directory_and_filename(path, destination=destination)
+        handle = await self.handle(path)
+        return await asynfs.write(directory, name, handle)
+
+    async def download_many(self, target, objects, destination=None):
+        """
+        Download selected files and/or directories as a ZIP archive.
+
+        .. warning::
+            The provided list of objects is not validated. Only existing files and directories
+            will be included in the resulting ZIP file.
+
+        :param str target:
+            Path to the cloud folder containing the files and directories to download.
+        :param list[str] objects:
+            List of file and/or directory names to include in the download.
+        :param str destination:
+            Optional. Path to the destination file or directory. If a directory is provided,
+            the original filename will be preserved. Defaults to the default download directory.
+        """
+        directory, name = commonfs.determine_directory_and_filename(target, objects, destination=destination, archive=True)
+        handle = await self.handle_many(target, *objects)
+        return await asynfs.write(directory, name, handle)
+
     async def listdir(self, path, depth=None, include_deleted=False):
         """
         List Directory
@@ -116,7 +147,7 @@ async def upload_file(self, path, destination):
         :param str destination: Remote path
         """
         with open(path, 'rb') as handle:
-            metadata = self._filesystem.properties(path)
+            metadata = commonfs.properties(path)
             response = await self.upload(metadata['name'], metadata['size'], destination, handle)
         return response
 

cterasdk/core/files/browser.py

@@ -3,7 +3,7 @@
 import cterasdk.settings
 from ...cio.core import CorePath
 from ...exceptions import CTERAException
-from ...lib import FileSystem
+from ...lib.storage import synfs, commonfs
 from ..base_command import BaseCommand
 from . import io
 
@@ -13,7 +13,6 @@ class FileBrowser(BaseCommand):
     def __init__(self, core):
         super().__init__(core)
         self._scope = f'/{self._core.context}/webdav'
-        self._filesystem = FileSystem.instance()
 
     def handle(self, path):
         """
@@ -42,24 +41,29 @@ def download(self, path, destination=None):
         :param str,optional destination:
          File destination, if it is a directory, the original filename will be kept, defaults to the default directory
         """
-        directory, name = self.determine_directory_and_filename(path, destination=destination)
+        directory, name = commonfs.determine_directory_and_filename(path, destination=destination)
         handle = self.handle(path)
-        return self._filesystem.save(directory, name, handle)
+        return synfs.write(directory, name, handle)
 
-    def download_as_zip(self, target, objects, destination=None):
+    def download_many(self, target, objects, destination=None):
         """
-        Download a list of files and/or directories from a cloud folder as a ZIP file
+        Download selected files and/or directories as a ZIP archive.
 
-        .. warning:: The list of files is not validated. The ZIP file will include only the existing  files and directories
+        .. warning::
+            The provided list of objects is not validated. Only existing files and directories
+            will be included in the resulting ZIP file.
 
-        :param str target: Path to the cloud directory
-        :param list[str] objects: List of files and/or directories in the cloud folder to download
-        :param str,optional destination:
-         File destination, if it is a directory, the original filename will be kept, defaults to the default directory
+        :param str target:
+            Path to the cloud folder containing the files and directories to download.
+        :param list[str] objects:
+            List of file and/or directory names to include in the download.
+        :param str destination:
+            Optional. Path to the destination file or directory. If a directory is provided,
+            the original filename will be preserved. Defaults to the default download directory.
         """
-        directory, name = self.determine_directory_and_filename(target, objects, destination=destination, archive=True)
+        directory, name = commonfs.determine_directory_and_filename(target, objects, destination=destination, archive=True)
         handle = self.handle_many(target, *objects)
-        return self._filesystem.save(directory, name, handle)
+        return synfs.write(directory, name, handle)
 
     def listdir(self, path, depth=None, include_deleted=False):
         """
@@ -120,31 +124,6 @@ def permalink(self, path):
             return contents[0].permalink
         raise FileNotFoundError('File not found.', path)
 
-    def determine_directory_and_filename(self, p, objects=None, destination=None, archive=False):
-        """
-        Determine location to save file.
-
-        :param str p: Path.
-        :param list[str],optional objects: List of files or folders
-        :param str,optional destination: Destination
-        :param bool,optional archive: Compressed archive
-        :returns: Directory and file name
-        :rtype: tuple[str]
-        """
-        directory, name = None, None
-        if destination:
-            directory, name = self._filesystem.split_file_directory(destination)
-        else:
-            directory = self._filesystem.downloads_directory()
-
-        if not name:
-            normalized = self.normalize(p)
-            if archive:
-                name = self._filesystem.compute_zip_file_name(normalized.absolute, objects)
-            else:
-                name = normalized.name
-        return directory, name
-
     def normalize(self, entries):
         return CorePath.instance(self._scope, entries)
 
@@ -171,7 +150,7 @@ def upload_file(self, path, destination):
         :param str destination: Remote path
         """
         with open(path, 'rb') as handle:
-            metadata = self._filesystem.properties(path)
+            metadata = commonfs.properties(path)
             response = self.upload(metadata['name'], metadata['size'], destination, handle)
         return response
 

cterasdk/core/ssl.py

@@ -2,18 +2,15 @@
 from zipfile import ZipFile
 
 from .base_command import BaseCommand
-from ..lib import FileSystem, X509Certificate, PrivateKey, TempfileServices, create_certificate_chain
+from ..lib import X509Certificate, PrivateKey, TempfileServices, create_certificate_chain
+from ..lib.storage import commonfs, synfs
 
 
 class SSL(BaseCommand):
     """
     Portal SSL Certificate APIs
     """
 
-    def __init__(self, portal):
-        super().__init__(portal)
-        self._filesystem = FileSystem.instance()
-
     def get(self):
         """
         Retrieve details of the current installed SSL certificate
@@ -39,14 +36,15 @@ def export(self, destination=None):
         :param str,optional destination:
          File destination, defaults to the default directory
         """
-        directory, filename = self._filesystem.generate_file_location(destination, 'certificate.zip')
+        directory, filename = commonfs.generate_file_destination(destination, 'certificate.zip')
         logging.getLogger('cterasdk.core').info('Exporting SSL certificate.')
         handle = self._core.ctera.handle('/preview/exportCertificate')
-        filepath = self._filesystem.save(directory, filename, handle)
+        filepath = synfs.write(directory, filename, handle)
         logging.getLogger('cterasdk.core').info('Exported SSL certificate. %s', {'filepath': filepath})
         return filepath
 
-    def create_zip_archive(self, private_key, *certificates):
+    @staticmethod
+    def create_zip_archive(private_key, *certificates):
         """
         Create a ZIP archive that can be imported to CTERA Portal
 
@@ -57,22 +55,22 @@ def create_zip_archive(self, private_key, *certificates):
 
         key_basename = 'private.key'
         key_object = PrivateKey.load_private_key(private_key)
-        key_filepath = FileSystem.join(tempdir, key_basename)
-        self._filesystem.write(key_filepath, key_object.pem_data)
+        key_filepath = commonfs.join(tempdir, key_basename)
+        synfs.overwrite(key_filepath, key_object.pem_data)
 
         cert_basename = 'certificate'
         certificates = [X509Certificate.load_certificate(certificate) for certificate in certificates]
         certificate_chain = create_certificate_chain(*certificates)
 
         certificate_chain_zip_archive = None
         if certificate_chain:
-            certificate_chain_zip_archive = FileSystem.join(tempdir, f'{cert_basename}.zip')
+            certificate_chain_zip_archive = commonfs.join(tempdir, f'{cert_basename}.zip')
             with ZipFile(certificate_chain_zip_archive, 'w') as zip_archive:
                 zip_archive.write(key_filepath, key_basename)
                 for idx, certificate in enumerate(certificate_chain):
                     filename = f'{cert_basename}{idx if idx > 0 else ""}.crt'
-                    filepath = FileSystem.join(tempdir, filename)
-                    self._filesystem.write(filepath, certificate.pem_data)
+                    filepath = commonfs.join(tempdir, filename)
+                    synfs.overwrite(filepath, certificate.pem_data)
                     zip_archive.write(filepath, filename)
 
         return certificate_chain_zip_archive
@@ -92,11 +90,11 @@ def import_from_chain(self, private_key, *certificates):
         :param str private_key: The PEM-encoded private key, or a path to the PEM-encoded private key file
         :param list[str] certificates: The PEM-encoded certificates, or a list of paths of the PEM-encoded certificate files
         """
-        zipflie = self.create_zip_archive(private_key, *certificates)
+        zipflie = SSL.create_zip_archive(private_key, *certificates)
         return self.import_from_zip(zipflie)
 
     def _import_certificate(self, zipfile):
-        self._filesystem.properties(zipfile)
+        commonfs.properties(zipfile)
         logging.getLogger('cterasdk.core').info('Uploading SSL certificate.')
         with open(zipfile, 'rb') as fd:
             response = self._core.api.form_data(

cterasdk/edge/config.py

@@ -5,17 +5,17 @@
 from ..exceptions import CTERAException
 from ..convert import fromxmlstr, toxmlstr
 from ..common import Device, delete_attrs
-from ..lib import FileSystem, TempfileServices
+from ..lib import TempfileServices
+from ..lib.storage import commonfs, synfs
 from .base_command import BaseCommand
 
 
+logger = logging.getLogger('cterasdk.edge')
+
+
 class Config(BaseCommand):
     """ Edge Filer General Configuration APIs """
 
-    def __init__(self, edge):
-        super().__init__(edge)
-        self._filesystem = FileSystem.instance()
-
     def get_location(self):
         """
         Get the location of the Edge Filer
@@ -31,7 +31,7 @@ def set_location(self, location):
         :param str location: New location to set
         :return str: The new location
         """
-        logging.getLogger('cterasdk.edge').info('Configuring device location. %s', {'location': location})
+        logger.info('Configuring device location. %s', {'location': location})
         return self._edge.api.put('/config/device/location', location)
 
     def get_hostname(self):
@@ -49,7 +49,7 @@ def set_hostname(self, hostname):
         :param str hostname: New hostname to set
         :return str: The new hostname
         """
-        logging.getLogger('cterasdk.edge').info('Configuring device hostname. %s', {'hostname': hostname})
+        logger.info('Configuring device hostname. %s', {'hostname': hostname})
         return self._edge.api.put('/config/device/hostname', hostname)
 
     def import_config(self, config, exclude=None):
@@ -64,19 +64,19 @@ def import_config(self, config, exclude=None):
         if isinstance(config, Device):
             database = copy.deepcopy(config)
         elif isinstance(config, str):
-            database = self.load_config(config)
+            database = Config.load_config(config)
 
         if exclude:
             delete_attrs(database, exclude)
 
-        path = self._filesystem.join(TempfileServices.mkdir(), f'{self._edge.session().address}.xml')
-        self._filesystem.write(path, toxmlstr(database, True).encode('utf-8'))
+        path = commonfs.join(TempfileServices.mkdir(), f'{self._edge.session().address}.xml')
+        synfs.overwrite(path, toxmlstr(database, True).encode('utf-8'))
 
         return self._import_configuration(path)
 
     def _import_configuration(self, path):
-        self._filesystem.properties(path)
-        logging.getLogger('cterasdk.edge').info('Importing Edge Filer configuration.')
+        commonfs.properties(path)
+        logger.info('Importing Edge Filer configuration.')
         with open(path, 'rb') as fd:
             response = self._edge.api.form_data(
                 '/config',
@@ -86,28 +86,29 @@ def _import_configuration(self, path):
                     config=fd
                 )
             )
-            logging.getLogger('cterasdk.edge').info('Imported Edge Filer configuration.')
+            logger.info('Imported Edge Filer configuration.')
         return response
 
-    def load_config(self, config):
+    @staticmethod
+    def load_config(config):
         """
         Load the Edge Filer configuration
 
         :param str config: A string or a path to the Edge Filer configuration file
         """
         data = None
-        if self._filesystem.exists(config):
-            logging.getLogger('cterasdk.edge').info('Reading the Edge Filer configuration from file. %s', {'path': config})
+        if commonfs.exists(config):
+            logger.info('Reading the Edge Filer configuration from file. %s', {'path': config})
             with open(config, 'r', encoding='utf-8') as f:
                 data = f.read()
         else:
             data = config
 
         database = fromxmlstr(data)
         if database:
-            logging.getLogger('cterasdk.edge').info('Completed parsing the Edge Filer configuration. %s', {'firmware': database.firmware})
+            logger.info('Completed parsing the Edge Filer configuration. %s', {'firmware': database.firmware})
             return database
-        logging.getLogger('cterasdk.edge').error("Failed parsing the Edge Filer's configuration.")
+        logger.error("Failed parsing the Edge Filer's configuration.")
         raise CTERAException("Failed parsing the Edge Filer's configuration")
 
     def export(self, destination=None):
@@ -118,11 +119,12 @@ def export(self, destination=None):
          File destination, defaults to the default directory
         """
         default_filename = self._edge.host() + datetime.now().strftime('_%Y-%m-%dT%H_%M_%S') + '.xml'
-        directory, filename = self._filesystem.generate_file_location(destination, default_filename)
-        logging.getLogger('cterasdk.edge').info('Exporting configuration. %s', {'host': self._edge.host()})
+        directory, filename = commonfs.generate_file_destination(destination, default_filename)
+        logger.info('Exporting configuration. %s', {'host': self._edge.host()})
         handle = self._edge.api.handle('/export')
-        filepath = FileSystem.instance().save(directory, filename, handle)
-        logging.getLogger('cterasdk.edge').info('Exported configuration. %s', {'filepath': filepath})
+        filepath = synfs.write(directory, filename, handle)
+        logger.info('Exported configuration. %s', {'filepath': filepath})
+        return filepath
 
     def is_wizard_enabled(self):
         """
@@ -145,5 +147,5 @@ def disable_wizard(self):
         return self._set_wizard(False)
 
     def _set_wizard(self, state):
-        logging.getLogger('cterasdk.edge').info('Disabling first time wizard')
+        logger.info('Disabling first time wizard')
         return self._edge.api.put('/config/gui/openFirstTimeWizard', state)