Adds simple separated for POC pydantic settings

core/libs/commonwealth/commonwealth/settings/bases/pydantic_base.py

@@ -0,0 +1,108 @@
+import abc
+import json
+import pathlib
+from typing import Any, ClassVar, Dict, List
+
+from loguru import logger
+from pydantic import BaseModel, ValidationError
+
+from commonwealth.settings.exceptions import (
+    BadAttributes,
+    BadSettingsClassNaming,
+    BadSettingsFile,
+    MigrationFail,
+    SettingsFromTheFuture,
+)
+
+
+class PydanticSettings(BaseModel):
+    VERSION: int = 0
+    STATIC_VERSION: ClassVar[int]
+
+    def __init__(self, **kwargs: Dict[str, Any]) -> None:
+        super().__init__(**kwargs)
+        direct_children: List[Any] = []
+        for child in type(self).mro():
+            if child == PydanticSettings:
+                break
+            if issubclass(child, PydanticSettings):
+                direct_children.append(child)
+        for child in reversed(direct_children):
+            try:
+                v = int("".join(filter(str.isdigit, child.__name__)))
+            except ValueError as e:
+                raise BadSettingsClassNaming(
+                    f"{child.__name__} is not a valid settings class name, valid names should contain as number. Eg: V1"
+                ) from e
+            self.VERSION = v
+            child.STATIC_VERSION = v  # type: ignore
+
+    @abc.abstractmethod
+    def migrate(self, data: Dict[str, Any]) -> None:
+        """Function used to migrate from previous settings version
+
+        Args:
+            data (dict): Data from the previous version settings
+        """
+        raise RuntimeError("Migrating the settings file does not appears to be possible.")
+
+    def load(self, file_path: pathlib.Path) -> None:
+        """Load settings from file
+
+        Args:
+            file_path (pathlib.Path): Path for settings file
+        """
+        if not file_path.exists():
+            raise RuntimeError(f"Settings file does not exist: {file_path}")
+
+        logger.debug(f"Loading settings from file: {file_path}")
+        with open(file_path, encoding="utf-8") as settings_file:
+            result = json.load(settings_file)
+
+            if "VERSION" not in result.keys():
+                raise BadSettingsFile(f"Settings file does not appears to contain a valid settings format: {result}")
+
+            version = result["VERSION"]
+
+            if version <= 0:
+                raise BadAttributes("Settings file contains invalid version number")
+
+            if version > self.VERSION:
+                raise SettingsFromTheFuture(
+                    f"Settings file comes from a future settings version: {version}, "
+                    f"latest supported: {self.VERSION}, tomorrow does not exist"
+                )
+
+            if version < self.VERSION:
+                self.migrate(result)
+                version = result["VERSION"]
+
+            if version != self.VERSION:
+                raise MigrationFail("Migrate chain failed to update to the latest settings version available")
+
+            # Copy new content to settings class
+            try:
+                new = self.parse_obj(result)
+                self.__dict__.update(new.__dict__)
+            except ValidationError as e:
+                raise BadSettingsFile(f"Settings file contains invalid data: {e}") from e
+
+    def save(self, file_path: pathlib.Path) -> None:
+        """Save settings to file
+
+        Args:
+            file_path (pathlib.Path): Path for the settings file
+        """
+        # Path for settings file does not exist, lets ensure that it does
+        parent_path = file_path.parent.absolute()
+        parent_path.mkdir(parents=True, exist_ok=True)
+
+        with open(file_path, "w", encoding="utf-8") as settings_file:
+            logger.debug(f"Saving settings on: {file_path}")
+            settings_file.write(self.json(indent=4))
+
+    def reset(self) -> None:
+        """Reset internal data to default values"""
+        logger.debug("Resetting settings")
+        new = self.__class__()
+        self.__dict__.update(new.__dict__)

core/libs/commonwealth/commonwealth/settings/bases/pykson_base.py

@@ -0,0 +1,101 @@
+import abc
+import json
+import pathlib
+from typing import Any, Dict
+
+import pykson  # type: ignore
+from loguru import logger
+from pykson import Field, Pykson
+
+from commonwealth.settings.exceptions import (
+    BadAttributes,
+    BadSettingsFile,
+    MigrationFail,
+    SettingsFromTheFuture,
+)
+
+
+class PyksonSettings(pykson.JsonObject):
+    """Base settings class for Pykson serializer"""
+
+    VERSION = pykson.IntegerField(default_value=0)
+
+    def __init__(self, *args: str, **kwargs: int) -> None:
+        # Make sure that all attributes are derivated from Pykson.Field
+        for key, item in type(self).__dict__.items():
+            # Remove default attributes and version tracker from validation
+            if key in ["__doc__", "__module__", "VERSION"]:
+                continue
+            if callable(item):
+                continue
+            assert isinstance(
+                item, Field
+            ), f"Class attributes must be from Pykson.Field or derivated: {type(item)}: {key}"
+        super().__init__(*args, **kwargs)
+
+    @abc.abstractmethod
+    def migrate(self, data: Dict[str, Any]) -> None:
+        """Function used to migrate from previous settings version
+
+        Args:
+            data (dict): Data from the previous version settings
+        """
+        raise RuntimeError("Migrating the settings file does not appears to be possible.")
+
+    def load(self, file_path: pathlib.Path) -> None:
+        """Load settings from file
+
+        Args:
+            file_path (pathlib.Path): Path for settings file
+        """
+        if not file_path.exists():
+            raise RuntimeError(f"Settings file does not exist: {file_path}")
+
+        logger.debug(f"Loading settings from file: {file_path}")
+        with open(file_path, encoding="utf-8") as settings_file:
+            result = json.load(settings_file)
+
+            if "VERSION" not in result.keys():
+                raise BadSettingsFile(f"Settings file does not appears to contain a valid settings format: {result}")
+
+            version = result["VERSION"]
+
+            if version <= 0:
+                raise BadAttributes("Settings file contains invalid version number")
+
+            if version > self.VERSION:
+                raise SettingsFromTheFuture(
+                    f"Settings file comes from a future settings version: {version}, "
+                    f"latest supported: {self.VERSION}, tomorrow does not exist"
+                )
+
+            if version < self.VERSION:
+                self.migrate(result)
+                version = result["VERSION"]
+
+            if version != self.VERSION:
+                raise MigrationFail("Migrate chain failed to update to the latest settings version available")
+
+            # Copy new content to settings class
+            new = Pykson().from_json(result, self.__class__)
+            self.__dict__.update(new.__dict__)
+
+    def save(self, file_path: pathlib.Path) -> None:
+        """Save settings to file
+
+        Args:
+            file_path (pathlib.Path): Path for the settings file
+        """
+        # Path for settings file does not exist, lets ensure that it does
+        parent_path = file_path.parent.absolute()
+        parent_path.mkdir(parents=True, exist_ok=True)
+
+        with open(file_path, "w", encoding="utf-8") as settings_file:
+            logger.debug(f"Saving settings on: {file_path}")
+            settings_file.write(Pykson().to_json(self))
+
+    def reset(self) -> None:
+        """Reset internal data to default values"""
+        logger.debug("Resetting settings")
+        new = self.__class__()
+        self.__dict__.update(new.__dict__)

core/libs/commonwealth/commonwealth/settings/exceptions.py

@@ -0,0 +1,18 @@
+class BadSettingsFile(ValueError):
+    """Settings file is not valid."""
+
+
+class SettingsFromTheFuture(ValueError):
+    """Settings file version is from a newer version of the service."""
+
+
+class MigrationFail(RuntimeError):
+    """Could not apply migration."""
+
+
+class BadAttributes(BadSettingsFile):
+    """Attributes on settings file are not valid."""
+
+
+class BadSettingsClassNaming(RuntimeError):
+    """Setting class in the inheritance chain have a name that is not valid."""

core/libs/commonwealth/commonwealth/settings/manager.py

@@ -1,125 +1,3 @@
-import pathlib
-import re
-from typing import Any, Optional, Type
+from commonwealth.settings.managers.pykson_manager import PyksonManager as Manager
 
-import appdirs
-from loguru import logger
-
-from commonwealth.settings.settings import BaseSettings, SettingsFromTheFuture
-
-
-class Manager:
-    SETTINGS_NAME_PREFIX = "settings-"
-
-    def __init__(
-        self,
-        project_name: str,
-        settings_type: Type[BaseSettings],
-        config_folder: Optional[pathlib.Path] = None,
-        load: bool = True,
-    ) -> None:
-        assert project_name, "project_name should be not empty"
-        assert issubclass(settings_type, BaseSettings), "settings_type should use BaseSettings as subclass"
-
-        self.project_name = project_name.lower()
-        self.config_folder = (
-            config_folder.joinpath(self.project_name)
-            if config_folder
-            else pathlib.Path(appdirs.user_config_dir(self.project_name))
-        )
-        self.config_folder.mkdir(parents=True, exist_ok=True)
-        self.settings_type = settings_type
-        self._settings = None
-        logger.debug(
-            f"Starting {project_name} settings with {settings_type.__name__}, configuration path: {config_folder}"
-        )
-        if load:
-            self.load()
-
-    @property
-    def settings(self) -> Any:
-        """Getter point for settings
-
-        Returns:
-            [Type[BaseSettings]]: The settings defined in the constructor
-        """
-        if not self._settings:
-            self.load()
-
-        return self._settings
-
-    @settings.setter
-    def settings(self, value: Any) -> None:
-        """Setter point for settings. Save settings for every change
-
-        Args:
-            value ([Type[BaseSettings]]): The settings defined in the constructor
-        """
-        if not self._settings:
-            self.load()
-
-        self._settings = value
-        self.save()
-
-    def settings_file_path(self) -> pathlib.Path:
-        """Return the settings file for the version specified in the constructor settings
-
-        Returns:
-            pathlib.Path: Path for the settings file
-        """
-        return self.config_folder.joinpath(f"{Manager.SETTINGS_NAME_PREFIX}{self.settings_type.VERSION}.json")
-
-    @staticmethod
-    def load_from_file(settings_type: Type[BaseSettings], file_path: pathlib.Path) -> Any:
-        """Load settings from a generic location and settings type
-
-        Args:
-            settings_type (BaseSettings): Settings type that inherits from BaseSettings.
-            file_path (pathlib.Path): Path for a valid settings file
-
-        Returns:
-            Any: The settings based on settings_type
-        """
-        assert issubclass(settings_type, BaseSettings), "settings_type should use BaseSettings as subclass"
-
-        settings_data = settings_type()
-
-        if file_path.exists():
-            settings_data.load(file_path)
-        else:
-            settings_data.save(file_path)
-
-        return settings_data
-
-    def save(self) -> None:
-        """Save settings"""
-        self.settings.save(self.settings_file_path())
-
-    def load(self) -> None:
-        """Load settings"""
-
-        def get_settings_version_from_filename(filename: pathlib.Path) -> int:
-            result = re.search(f"{Manager.SETTINGS_NAME_PREFIX}(\\d+)", filename.name)
-            assert result
-            assert len(result.groups()) == 1
-            return int(result.group(1))
-
-        # Get all possible settings candidates and sort it by version
-        valid_files = [
-            possible_file
-            for possible_file in self.config_folder.iterdir()
-            if possible_file.name.startswith(Manager.SETTINGS_NAME_PREFIX)
-        ]
-        valid_files.sort(key=get_settings_version_from_filename, reverse=True)
-
-        logger.debug(f"Found possible candidates for settings source: {valid_files}")
-        for valid_file in valid_files:
-            logger.debug(f"Checking {valid_file} for settings")
-            try:
-                self._settings = Manager.load_from_file(self.settings_type, valid_file)
-                logger.debug(f"Using {valid_file} as settings source")
-                return
-            except SettingsFromTheFuture as exception:
-                logger.debug("Invalid settings, going to try another file:", exception)
-
-        self._settings = Manager.load_from_file(self.settings_type, self.settings_file_path())
+__all__ = ["Manager"]