Add new feature Config class

.github/workflows/python-check.yml

@@ -25,7 +25,7 @@ jobs:
           pip install -r requirements.txt
       - name: Runt tests
         run: |
-          export PYTHONPATH=src
+          export PYTHONPATH=.
           python -m unittest -v ./tests/test_daemon.py
       - name: Lint
         run: |

README.md

@@ -6,6 +6,7 @@ Small boiler plate with tools for multithreading.
 Minibone is a set of valuable classes for:
 
 - __Daemon__: for multithreading tasks
+- __Config__: To handle configuration setting
 - Among others (I will add more later)
 
 It will be deployed to PyPi when a new release is created
@@ -43,6 +44,20 @@ Check [sample_clock.py](https://github.com/erromu/minibone/blob/main/src/minibon
 
 Check [sample_clock_callback.py](https://github.com/erromu/minibone/blob/main/src/minibone/sample_clock_callback.py) for a sample
 
+### Config
+
+Allows to handle configuration settings in memory and persists them into toml/yaml/json format
+
+> from minibone.config import Config
+>
+> cfg = Config(settings={"listen": "localhost", "port": 80}, filepath="config.toml")
+> 
+> cfg.add("debug", True)
+>
+> cfg.to_toml()
+>
+> cfg2 = Config.from_toml("config.toml")
+
 ## Contribution
 
 - Feel free to clone this repository, and send any pull requests.

pyproject.toml

@@ -4,7 +4,7 @@
 
 [project]
 name = "minibone"
-version = "0.0.4"
+version = "0.1.0"
 description = "Small boiler plate with tools for multithreading."
 keywords = ["multithreading", "task", "job", "background"]
 authors = [
@@ -24,7 +24,7 @@ classifiers = [
     "Topic :: Software Development :: Libraries",
 ]
 requires-python = ">=3.0"
-dependencies = []
+dependencies = ["tomlkit", "pyyaml"]
 
 [project.urls]
 "Homepage" = "https://github.com/erromu/minibone/"

requirements.txt

@@ -0,0 +1,2 @@
+pyyaml>=6.0.2
+tomlkit>=0.13.2

src/minibone/config.py

@@ -0,0 +1,202 @@
+import json
+import logging
+import re
+from datetime import date, datetime, time
+from enum import Enum
+from pathlib import Path
+
+import tomlkit
+import yaml
+
+
+class FORMAT(Enum):
+    TOML = "TOML"
+    YAML = "YAML"
+    JSON = "JSON"
+
+
+class Config(dict):
+    """Class to have settings in memory or in a configuration file"""
+
+    @classmethod
+    def from_toml(cls, filepath: str, defaults: dict = None):
+        """Load a toml configuration file and return a Config instance
+
+        Arguments
+        ---------
+        filepath:   str     The filepath of the file to load
+        defaults:   dict    A dictionary with default settings.
+                            Values from the file will expand/replace defaults
+        """
+        assert isinstance(filepath, str) and len(filepath) > 0
+        logger = logging.getLogger(__class__.__name__)
+
+        settings = {}
+
+        try:
+            file = "{path}".format(path=filepath)
+            with open(file, "rt", encoding="utf-8") as f:
+                settings = tomlkit.load(f)
+        except Exception as e:
+            logger.error("from_toml error loading %s. %s", filepath, e)
+
+        return Config(cls.merge(defaults, settings), filepath)
+
+    @classmethod
+    def from_yaml(cls, filepath: str, defaults: dict = None):
+        """Load a yaml configuration file and return a Config instance
+
+        Arguments
+        ---------
+        filepath:   str     The filepath of the file to load
+        defaults:   dict    A dictionary with default settings.
+                            Values from the file will expand/replace defaults
+        """
+        assert isinstance(filepath, str) and len(filepath) > 0
+        logger = logging.getLogger(__class__.__name__)
+
+        settings = {}
+
+        try:
+            file = "{path}".format(path=filepath)
+            with open(file, "rt", encoding="utf-8") as f:
+                settings = yaml.safe_load(f)
+        except Exception as e:
+            logger.error("from_yaml error loading %s. %s", filepath, e)
+
+        return Config(cls.merge(defaults, settings), filepath)
+
+    @classmethod
+    def from_json(cls, filepath: str, defaults: dict = None):
+        """Load a json configuration file and return a Config instance
+
+        Arguments
+        ---------
+        filepath:   str     The filepath of the file to load
+        defaults:   dict    A dictionary with default settings.
+                            Values from the file will expand/replace defaults
+        """
+        assert isinstance(filepath, str) and len(filepath) > 0
+        logger = logging.getLogger(__class__.__name__)
+
+        settings = {}
+
+        try:
+            file = "{path}".format(path=filepath)
+            with open(file, "rt", encoding="utf-8") as f:
+                settings = json.load(f)
+        except Exception as e:
+            logger.error("from_json error loading %s. %s", filepath, e)
+
+        return Config(cls.merge(defaults, settings), filepath)
+
+    @classmethod
+    def merge(cls, defaults: dict = None, settings: dict = None) -> dict:
+        """Merge settings into defaults (replace/expand defaults)
+
+        Arguments
+        ---------
+        defaults:   dict    The default settings
+        settings:   dict    The settings to expand/replace into defaults
+        """
+        assert not defaults or isinstance(defaults, dict)
+        assert not settings or isinstance(settings, dict)
+
+        if not defaults:
+            defaults = {}
+        if not settings:
+            settings = {}
+
+        return defaults | settings
+
+    def __init__(self, settings: dict = {}, filepath: str = None):
+        """
+        Arguments
+        ---------
+        settings:   dict    A dictionary of settings
+                            Each key in the dictionary must start with lowercase a-z
+                            and only ASCII characters are allowed in the name [a-ZA-Z_0-9]
+
+
+        filepath:   str     Full filepath of the file to store settings in
+        """
+        assert isinstance(settings, dict)
+        assert not filepath or isinstance(filepath, str)
+        self._logger = logging.getLogger(__class__.__name__)
+
+        self.filepath = filepath
+
+        for key, value in settings.items():
+            self.add(key, value)
+
+    def _parent_exits(self):
+        """create the parent directory if it does not exits"""
+        file = Path(self.filepath)
+        parent = Path(file.parent) if not file.exists() else None
+        if parent and not parent.exists():
+            parent.mkdir(exist_ok=True, parents=True)
+
+    def to_toml(self):
+        """Save settings to file in toml format"""
+        if not self.filepath:
+            self._logger.error("Not filepath defined for to_toml. Aborting")
+            return
+
+        try:
+            self._parent_exits()
+            with open(self.filepath, "wt", encoding="utf-8") as f:
+                tomlkit.dump(self.copy(), f)
+
+        except Exception as e:
+            self._logger.error("to_toml error %s. %s", self.filepath, e)
+
+    def to_yaml(self):
+        """Save settings to file in yaml format"""
+        if not self.filepath:
+            self._logger.error("Not filepath defined for to_yaml. Aborting")
+            return
+
+        try:
+            self._parent_exits()
+            with open(self.filepath, "wt", encoding="utf-8") as f:
+                yaml.dump(self.copy(), f)
+
+        except Exception as e:
+            self._logger.error("to_yaml error %s. %s", self.filepath, e)
+
+    def to_json(self):
+        """Save settings to file in json format"""
+        if not self.filepath:
+            self._logger.error("Not filepath defined for to_json. Aborting")
+            return
+
+        try:
+            self._parent_exits()
+            with open(self.filepath, "wt", encoding="utf-8") as f:
+                json.dump(self.copy(), f)
+
+        except Exception as e:
+            self._logger.error("to_json error %s. %s", self.filepath, e)
+
+    def add(self, key: str, value):
+        """Add/set a setting
+        Arguments
+        ---------
+        key:    str         A str valid key to name this setting.
+                            The key name must star with a lowercase [a-z], and contain ASCII characters only
+
+        value   object      Value of the setting.  The only allowed values are:
+                            str, int, float, list, dict, bool, datetime, date, time
+        """
+        assert isinstance(key, str) and re.match("[a-z]\w", key)
+        assert isinstance(value, (str, int, float, list, dict, bool, datetime, date, time))
+
+        self[key] = value
+
+    def remove(self, key: str):
+        """Remove a setting from this configuration
+        Arguments
+        ---------
+        key:    str         The key of the setting to remove
+        """
+        self.pop(key, None)

src/minibone/daemon.py

@@ -92,8 +92,8 @@ def __init__(
     def on_process(self):
         """Process to be called on each interation.
 
-        If a callback was  added, then it will be called instead.
-        When subclasing Daemon, verwrite this method to add your logic to be run
+        If a callback was added, then it will be called instead.
+        When subclasing Daemon, rewrite this method to add your logic to be run
         """
         pass
 

tests/test_config.py

@@ -0,0 +1,48 @@
+import unittest
+from pathlib import Path
+
+from src.minibone.config import Config
+
+
+class TestConfig(unittest.TestCase):
+    def test_settings(self):
+        settings = {"setting1": "value1", "setting2": 2, "setting3": True}
+        cfg = Config(settings=settings, filepath=None)
+
+        self.assertEqual(cfg.get("setting1", None), "value1")
+        self.assertEqual(cfg.get("setting10", None), None)
+        self.assertEqual(cfg.get("setting2", None), 2)
+        self.assertEqual(cfg.get("setting3", None), True)
+
+        cfg.remove("setting1")
+        cfg.add("setting3", False)
+
+        self.assertEqual(cfg.get("setting1", None), None)
+        self.assertEqual(cfg.get("setting3", None), False)
+
+        self.assertEqual(cfg.merge({}, {}), {})
+        self.assertEqual(cfg.merge(defaults={"x": 1}), {"x": 1})
+        self.assertEqual(cfg.merge(settings={"x": 1}), {"x": 1})
+        self.assertEqual(cfg.merge(defaults={"x": 1}, settings={"y": 2}), {"x": 1, "y": 2})
+        self.assertEqual(cfg.merge(defaults={"z": 1}, settings={"z": 4}), {"z": 4})
+
+        cfgs = []
+        files = ["config.toml", "config.yaml", "config.json"]
+        for file in files:
+            cfgs.append(Config(settings=cfg, filepath=file))
+
+        cfgs[0].to_toml()
+        cfgs[1].to_yaml()
+        cfgs[2].to_json()
+
+        self.assertEqual(cfgs[0].from_toml(cfgs[0].filepath), cfg)
+        self.assertEqual(cfgs[1].from_yaml(cfgs[1].filepath), cfg)
+        self.assertEqual(cfgs[2].from_json(cfgs[2].filepath), cfg)
+
+        for file in files:
+            p = Path(file)
+            p.unlink(missing_ok=True)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_daemon.py

@@ -2,7 +2,7 @@
 import unittest
 from threading import Lock
 
-from minibone.daemon import Daemon
+from src.minibone.daemon import Daemon
 
 
 class DaemonSubClass(Daemon):