feat: Abstract SamConfig

samcli/commands/deploy/guided_config.py

@@ -7,6 +7,7 @@
 
 from samcli.cli.context import get_cmd_names
 from samcli.commands.deploy.exceptions import GuidedDeployFailedError
+from samcli.lib.config.exceptions import SamConfigFileReadException
 from samcli.lib.config.samconfig import DEFAULT_CONFIG_FILE_NAME, DEFAULT_ENV, SamConfig
 
 
@@ -26,13 +27,17 @@ def get_config_ctx(self, config_file=None):
         return ctx, samconfig
 
     def read_config_showcase(self, config_file=None):
-        _, samconfig = self.get_config_ctx(config_file)
-
-        status = "Found" if samconfig.exists() else "Not found"
         msg = (
             "Syntax invalid in samconfig.toml; save values "
             "through sam deploy --guided to overwrite file with a valid set of values."
         )
+        try:
+            _, samconfig = self.get_config_ctx(config_file)
+        except SamConfigFileReadException:
+            raise GuidedDeployFailedError(msg)
+
+        status = "Found" if samconfig.exists() else "Not found"
+
         config_sanity = samconfig.sanity_check()
         click.secho("\nConfiguring SAM deploy\n======================", fg="yellow")
         click.echo(f"\n\tLooking for config file [{config_file}] :  {status}")

samcli/lib/config/exceptions.py

@@ -4,4 +4,12 @@
 
 
 class SamConfigVersionException(Exception):
-    pass
+    """Exception for the `samconfig` file being not present or in unrecognized format"""
+
+
+class FileParseException(Exception):
+    """Exception when a file is incorrectly parsed by a FileManager object."""
+
+
+class SamConfigFileReadException(Exception):
+    """Exception when a `samconfig` file is read incorrectly."""

samcli/lib/config/file_manager.py

@@ -0,0 +1,161 @@
+"""
+Class to represent the parsing of different file types into Python objects.
+"""
+
+
+import logging
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Any
+
+import tomlkit
+
+from samcli.lib.config.exceptions import FileParseException
+
+LOG = logging.getLogger(__name__)
+COMMENT_KEY = "__comment__"
+
+
+class FileManager(ABC):
+    """
+    Abstract class to be overridden by file managers for specific file extensions.
+    """
+
+    @staticmethod
+    @abstractmethod
+    def read(filepath: Path) -> Any:
+        """
+        Read a file at a given path.
+
+        Parameters
+        ----------
+        filepath: Path
+            The Path object that points to the file to be read.
+
+        Returns
+        -------
+        Any
+            A dictionary-like representation of the contents at the filepath location, along with a specialized
+            representation of the file that was read, if there is a specialization of it.
+        """
+        raise NotImplementedError("Read method not implemented.")
+
+    @staticmethod
+    @abstractmethod
+    def write(document: dict, filepath: Path):
+        """
+        Write a dictionary or dictionary-like object to a given file.
+
+        Parameters
+        ----------
+        document: dict
+            The object to write.
+        filepath: Path
+            The final location for the document to be written.
+        """
+        raise NotImplementedError("Write method not implemented.")
+
+    @staticmethod
+    @abstractmethod
+    def put_comment(document: Any, comment: str) -> Any:
+        """
+        Put a comment in a document object.
+
+        Parameters
+        ----------
+        document: Any
+            The object to write
+        comment: str
+            The comment to include in the document.
+
+        Returns
+        -------
+        Any
+            The new document, with the comment added to it.
+        """
+        raise NotImplementedError("Put comment method not implemented.")
+
+
+class TomlFileManager(FileManager):
+    """
+    Static class to read and write toml files.
+    """
+
+    file_format = "TOML"
+
+    @staticmethod
+    def read(filepath: Path) -> Any:
+        """
+        Read a TOML file at the given path.
+
+        Parameters
+        ----------
+        filepath: Path
+            The Path object that points to the file to be read.
+
+        Returns
+        -------
+        Any
+            A dictionary-like tomlkit.TOMLDocument object, which represents the contents of the TOML file at the
+            provided location.
+        """
+        toml_doc = tomlkit.document()
+        try:
+            txt = filepath.read_text()
+            toml_doc = tomlkit.loads(txt)
+        except OSError as e:
+            LOG.debug(f"OSError occurred while reading {TomlFileManager.file_format} file: {str(e)}")
+        except tomlkit.exceptions.TOMLKitError as e:
+            raise FileParseException(e) from e
+
+        return toml_doc
+
+    @staticmethod
+    def write(document: dict, filepath: Path):
+        """
+        Write the contents of a dictionary or tomlkit.TOMLDocument to a TOML file at the provided location.
+
+        Parameters
+        ----------
+        document: dict
+            The object to write.
+        filepath: Path
+            The final location for the TOML file to be written.
+        """
+        if not document:
+            LOG.debug("Nothing for TomlFileManager to write.")
+            return
+
+        toml_document = TomlFileManager._to_toml(document)
+
+        if toml_document.get(COMMENT_KEY, None):  # Remove dunder comments that may be residue from other formats
+            toml_document.add(tomlkit.comment(toml_document[COMMENT_KEY]))
+            toml_document.pop(COMMENT_KEY)
+
+        filepath.write_text(tomlkit.dumps(toml_document))
+
+    @staticmethod
+    def put_comment(document: dict, comment: str) -> Any:
+        """
+        Put a comment in a document object.
+
+        Parameters
+        ----------
+        document: Any
+            The tomlkit.TOMLDocument object to write
+        comment: str
+            The comment to include in the document.
+
+        Returns
+        -------
+        Any
+            The new TOMLDocument, with the comment added to it.
+        """
+        document = TomlFileManager._to_toml(document)
+        document.add(tomlkit.comment(comment))
+        return document
+
+    @staticmethod
+    def _to_toml(document: dict) -> tomlkit.TOMLDocument:
+        """Ensure that a dictionary-like object is a TOMLDocument."""
+        return tomlkit.parse(tomlkit.dumps(document))

samcli/lib/config/samconfig.py

@@ -5,27 +5,28 @@
 import logging
 import os
 from pathlib import Path
-from typing import Any, Iterable
+from typing import Any, Dict, Iterable, Type
 
-import tomlkit
-
-from samcli.lib.config.exceptions import SamConfigVersionException
+from samcli.lib.config.exceptions import FileParseException, SamConfigFileReadException, SamConfigVersionException
+from samcli.lib.config.file_manager import FileManager, TomlFileManager
 from samcli.lib.config.version import SAM_CONFIG_VERSION, VERSION_KEY
 
 LOG = logging.getLogger(__name__)
 
-DEFAULT_CONFIG_FILE_EXTENSION = "toml"
-DEFAULT_CONFIG_FILE_NAME = f"samconfig.{DEFAULT_CONFIG_FILE_EXTENSION}"
+DEFAULT_CONFIG_FILE_EXTENSION = ".toml"
+DEFAULT_CONFIG_FILE_NAME = f"samconfig{DEFAULT_CONFIG_FILE_EXTENSION}"
 DEFAULT_ENV = "default"
 DEFAULT_GLOBAL_CMDNAME = "global"
 
 
 class SamConfig:
     """
-    Class to interface with `samconfig.toml` file.
+    Class to represent `samconfig` config options.
     """
 
-    document = None
+    FILE_MANAGER_MAPPER: Dict[str, Type[FileManager]] = {
+        ".toml": TomlFileManager,
+    }
 
     def __init__(self, config_dir, filename=None):
         """
@@ -39,11 +40,21 @@ def __init__(self, config_dir, filename=None):
             Optional. Name of the configuration file. It is recommended to stick with default so in the future we
             could automatically support auto-resolving multiple config files within same directory.
         """
+        self.document = {}
         self.filepath = Path(config_dir, filename or DEFAULT_CONFIG_FILE_NAME)
+        self.file_manager = self.FILE_MANAGER_MAPPER.get(self.filepath.suffix, None)
+        if not self.file_manager:
+            LOG.warning(
+                f"The config file extension '{self.filepath.suffix}' is not supported. "
+                f"Supported formats are: [{'|'.join(self.FILE_MANAGER_MAPPER.keys())}]"
+            )
+            raise SamConfigFileReadException(
+                f"The config file {self.filepath} uses an unsupported extension, and cannot be read."
+            )
+        self._read()
 
     def get_stage_configuration_names(self):
-        self._read()
-        if isinstance(self.document, dict):
+        if self.document:
             return [stage for stage, value in self.document.items() if isinstance(value, dict)]
         return []
 
@@ -69,23 +80,19 @@ def get_all(self, cmd_names, section, env=DEFAULT_ENV):
         ------
         KeyError
             If the config file does *not* have the specific section
-
-        tomlkit.exceptions.TOMLKitError
-            If the configuration file is invalid
         """
 
         env = env or DEFAULT_ENV
 
-        self._read()
-        if isinstance(self.document, dict):
-            toml_content = self.document.get(env, {})
-            params = toml_content.get(self._to_key(cmd_names), {}).get(section, {})
-            if DEFAULT_GLOBAL_CMDNAME in toml_content:
-                global_params = toml_content.get(DEFAULT_GLOBAL_CMDNAME, {}).get(section, {})
-                global_params.update(params.copy())
-                params = global_params.copy()
-            return params
-        return {}
+        self.document = self._read()
+
+        config_content = self.document.get(env, {})
+        params = config_content.get(self._to_key(cmd_names), {}).get(section, {})
+        if DEFAULT_GLOBAL_CMDNAME in config_content:
+            global_params = config_content.get(DEFAULT_GLOBAL_CMDNAME, {}).get(section, {})
+            global_params.update(params.copy())
+            params = global_params.copy()
+        return params
 
     def put(self, cmd_names, section, key, value, env=DEFAULT_ENV):
         """
@@ -102,20 +109,10 @@ def put(self, cmd_names, section, key, value, env=DEFAULT_ENV):
         key : str
             Key to write the data under
         value : Any
-            Value to write. Could be any of the supported TOML types.
+            Value to write. Could be any of the supported types.
         env : str
             Optional, Name of the environment
-
-        Raises
-        ------
-        tomlkit.exceptions.TOMLKitError
-            If the data is invalid
         """
-
-        if self.document is None:
-            # Checking for None here since a TOMLDocument can include a
-            # 'body' property but still be falsy without a 'value' property
-            self._read()
         # Empty document prepare the initial structure.
         # self.document is a nested dict, we need to check each layer and add new tables, otherwise duplicated key
         # in parent layer will override the whole child layer
@@ -144,20 +141,12 @@ def put_comment(self, comment):
         comment: str
             A comment to write to the samconfg file
         """
-        if self.document is None:
-            self._read()
 
-        self.document.add(tomlkit.comment(comment))
+        self.document = self.file_manager.put_comment(self.document, comment)
 
     def flush(self):
         """
         Write the data back to file
-
-        Raises
-        ------
-        tomlkit.exceptions.TOMLKitError
-            If the data is invalid
-
         """
         self._write()
 
@@ -167,7 +156,7 @@ def sanity_check(self):
         """
         try:
             self._read()
-        except tomlkit.exceptions.TOMLKitError:
+        except SamConfigFileReadException:
             return False
         else:
             return True
@@ -196,13 +185,10 @@ def config_dir(template_file_path=None):
     def _read(self):
         if not self.document:
             try:
-                txt = self.filepath.read_text()
-                self.document = tomlkit.loads(txt)
-                self._version_sanity_check(self._version())
-            except OSError:
-                self.document = tomlkit.document()
-
-        if self.document.body:
+                self.document = self.file_manager.read(self.filepath)
+            except FileParseException as e:
+                raise SamConfigFileReadException(e) from e
+        if self.document:
             self._version_sanity_check(self._version())
         return self.document
 
@@ -213,12 +199,9 @@ def _write(self):
         self._ensure_exists()
 
         current_version = self._version() if self._version() else SAM_CONFIG_VERSION
-        try:
-            self.document.add(VERSION_KEY, current_version)
-        except tomlkit.exceptions.KeyAlreadyPresent:
-            # NOTE(TheSriram): Do not attempt to re-write an existing version
-            pass
-        self.filepath.write_text(tomlkit.dumps(self.document))
+        self.document.update({VERSION_KEY: current_version})
+
+        self.file_manager.write(self.document, self.filepath)
 
     def _version(self):
         return self.document.get(VERSION_KEY, None)