Initial implementation of schema and fields (#32)

Part of #10, #18, #26.

pyproject.toml

@@ -39,4 +39,4 @@ exclude_lines = [
 line_length = 88
 force_single_line = true
 known_first_party = ["resconfig"]
-known_third_party = ["pytest", "setuptools"]
+known_third_party = ["dateutil", "pytest", "setuptools"]

setup.py

@@ -31,7 +31,7 @@ def meta(category, fpath="src/resconfig/__init__.py"):
 readme = fread("README.rst")
 
 
-requires = []
+requires = ["python-dateutil>=2.8.1"]
 toml_requires = ["toml>=0.10.0"]
 yaml_requires = ["PyYAML>=5.3.1"]
 

src/resconfig/fields.py

@@ -0,0 +1,120 @@
+from datetime import datetime
+from datetime import timezone
+from functools import wraps
+
+from dateutil.parser import parse as dtparse
+
+from .ondict import ONDict
+
+
+def checktype(f):
+    @wraps(f)
+    def deco(cls, value):
+        if not isinstance(value, cls.ftype):
+            raise TypeError(f"{value} is not {cls.ftype}")
+        return f(cls, value)
+
+    return deco
+
+
+class Nullable:
+    default = None
+
+    @classmethod
+    def from_obj(cls, value):
+        if value is None or value == "null":
+            return None
+        return super().from_obj(value)
+
+    @classmethod
+    def to_str(cls, value):
+        if value is None:
+            return "null"
+        return super().to_str(value)
+
+
+class Field:
+    ftype = None
+    default = None
+
+    def __init__(self, value=None, doc=None, **kwargs):
+        super().__init__(**kwargs)
+        self.value = self.default if value is None else self.from_obj(value)
+        self.doc = doc
+
+    @classmethod
+    def from_obj(cls, value):
+        return cls.ftype(value)
+
+    @classmethod
+    @checktype
+    def to_str(cls, value):
+        return str(value)
+
+
+class Bool(Field):
+    ftype = bool
+    default = False
+
+    @classmethod
+    @checktype
+    def to_str(cls, value):
+        return "true" if value else "false"
+
+
+class Datetime(Field):
+    ftype = datetime
+    default = datetime.fromtimestamp(0, timezone.utc)
+
+    @classmethod
+    def from_obj(cls, value):
+        if isinstance(value, datetime):
+            pass
+        elif isinstance(value, str):
+            value = dtparse(value)
+        elif isinstance(value, (float, int)):
+            value = datetime.fromtimestamp(value, timezone.utc)
+        else:
+            raise ValueError(f"invalid value for datetime: {value!r}")
+        return value
+
+    @classmethod
+    @checktype
+    def to_str(cls, value):
+        return value.isoformat()
+
+
+class Float(Field):
+    ftype = float
+    default = 0.0
+
+
+class Int(Field):
+    ftype = int
+    default = 0
+
+
+class Str(Field):
+    ftype = str
+    default = ""
+
+
+# fmt: off
+class NullableBool(Nullable, Bool): pass
+class NullableDatetime(Nullable, Datetime): pass
+class NullableFloat(Nullable, Float): pass
+class NullableInt(Nullable, Int): pass
+class NullableStr(Nullable, Str): pass
+# fmt: on
+
+
+def extract_values(d: ONDict) -> ONDict:
+    """Make a ONDict with only default values and no other type info."""
+    valueonly = ONDict()
+    valueonly._create = True
+    for key in d.allkeys():
+        v = d[key]
+        if isinstance(v, Field):
+            v = v.value
+        valueonly[key] = v
+    return valueonly

src/resconfig/io/ini.py

@@ -1,48 +1,60 @@
 from collections.abc import MutableMapping
 from configparser import ConfigParser
 
+from .. import fields
+from ..ondict import ONDict
 from ..typing import IO
+from ..typing import Any
+from ..typing import Optional
+from .utils import escape_dot
+from .utils import unescape_dots_in_keys
 
 
-def load(f: IO) -> dict:
-    parser = ConfigParser()
-    parser.read_file(f)
-    dic = {}
-    for section in parser.sections():
-        ref = dic.setdefault(_escape_dot(section), {})
-        for option in parser[section]:
-            ref[_escape_dot(option)] = parser[section][option]
-    return dic
-
-
-def dump(content: dict, f: IO):
+def dump(content: ONDict, f: IO, schema: Optional[ONDict] = None):
     if _depth(content) > 2:
         raise ValueError("INI config does not allow nested options")
+    schema = schema or {}
 
-    _unescape_all_keys(content)
+    con = ONDict()
+    con._create = True
+    for key in list(content.allkeys()):
+        con[key] = _dumpobj(content[key], schema.get(key))
 
+    con = con.asdict()
+
+    unescape_dots_in_keys(con)
     parser = ConfigParser()
-    parser.read_dict(content)
+    parser.read_dict(con)
     parser.write(f)
 
 
-def _escape_dot(key: str) -> str:
-    return key.replace(".", r"\.")
+def _dumpobj(value, field) -> Any:
+    if isinstance(field, fields.Field):
+        value = field.to_str(value)
+    return value
 
 
-def _unescape_dot(key: str) -> str:
-    return key.replace(r"\.", ".")
+def load(f: IO, schema: Optional[ONDict] = None) -> ONDict:
+    schema = schema or {}
+    parser = ConfigParser()
+    parser.read_file(f)
+    conf = ONDict()
+    conf._create = True
+    for section in parser.sections():
+        section_key = escape_dot(section)
+        ref = conf.setdefault(section_key, ONDict())
+        for option in parser[section]:
+            option_key = escape_dot(option)
+            ref[option_key] = _loadobj(
+                schema.get((section_key, option_key)), parser[section][option]
+            )
+    return conf
 
 
-def _unescape_all_keys(d: dict):
-    if not isinstance(d, MutableMapping):
-        return
-    for k in list(d.keys()):
-        _unescape_all_keys(d[k])
-        uk = _unescape_dot(k)
-        if uk != k:
-            d[uk] = d[k]
-            del d[k]
+def _loadobj(field, value) -> Any:
+    if isinstance(field, fields.Field):
+        value = field.from_obj(value)
+    return value
 
 
 def _depth(d: dict, depth: int = 0) -> int:

src/resconfig/io/io.py

@@ -1,5 +1,3 @@
-from copy import deepcopy
-
 from ..ondict import ONDict
 from ..typing import FilePath
 from ..typing import List
@@ -12,11 +10,9 @@
 from .utils import ensure_path
 
 
-def _read_as_dict(filename: ConfigPath) -> ONDict:
-    return ONDict(filename.load())
-
-
-def read_from_files_as_dict(paths: List[FilePath], merge: bool = False) -> ONDict:
+def read_from_files_as_dict(
+    paths: List[FilePath], merge: bool = False, schema=None
+) -> ONDict:
     """Read the config from a file(s) as an :class:`ONDict`.
 
     How the config is constructed depends on the ``merge`` flag. If :obj:`True`, the
@@ -39,7 +35,7 @@ def read_from_files_as_dict(paths: List[FilePath], merge: bool = False) -> ONDic
         if path.is_file():
             if not isinstance(path, ConfigPath):
                 path = ConfigPath.from_extension(path)
-            content = _read_as_dict(path)
+            content = path.load(schema)
             d.merge(content)
             if not merge:
                 break
@@ -62,10 +58,10 @@ def update_from_files(self, paths: List[FilePath], merge=False):
             paths: A list of config file paths.
             merge: The flag for the merge mode; see the function description.
         """
-        self.update(read_from_files_as_dict(paths, merge))
+        self.update(read_from_files_as_dict(paths, merge, self._default))
 
     def __update_from_file(self, filename: ConfigPath):
-        self.update(_read_as_dict(filename))
+        self.update(filename.load(self._default))
 
     def update_from_file(self, filename: FilePath):
         """Update config from the file.
@@ -93,9 +89,7 @@ def update_from_yaml(self, filename: FilePath):
         self.__update_from_file(YAMLPath(filename))
 
     def __save(self, filename: ConfigPath):
-        d = deepcopy(self._conf)
-        self._schema.unapply_all(d)
-        filename.dump(d)
+        filename.dump(self._conf, schema=self._default)
 
     @experimental
     def save_to_file(self, filename: FilePath):

src/resconfig/io/json.py

@@ -1,15 +1,64 @@
+from collections.abc import MutableMapping
 from json import dump as _dump
 from json import load as _load
 from json.decoder import JSONDecodeError
+from logging import getLogger
 
+from .. import fields
+from ..ondict import ONDict
+from ..typing import IO
+from ..typing import Any
+from ..typing import Optional
+from ..typing import Union
+from .utils import escape_dot
 
-def dump(content, f):
-    return _dump(content, f)
+log = getLogger(__name__)
 
 
-def load(f):
+def dump(content: ONDict, f: IO, schema: Optional[ONDict] = None):
+    schema = schema or {}
+    con = ONDict()
+    con._create = True
+    for key in list(content.allkeys()):
+        con[key] = _dumpobj(content[key], schema.get(key))
+    _dump(con.asdict(), f)
+
+
+def _dumpobj(value, field) -> Any:
+    if isinstance(field, fields.Field):
+        if isinstance(field, (fields.Bool, fields.Float, fields.Int, fields.Str)):
+            pass
+        elif isinstance(field, (fields.Datetime,)):
+            value = field.to_str(value)
+        else:
+            value = field.to_str(value)
+    return value
+
+
+def load(f: IO, schema: Optional[ONDict] = None) -> ONDict:
     try:
         content = _load(f)
     except JSONDecodeError:
+        log.exception("Load error")
         content = {}
-    return content
+
+    def _walk(d, schema):
+        if not isinstance(d, MutableMapping):
+            return _loadobj(schema, d)
+        for key in list(d.keys()):
+            ekey = escape_dot(key)
+            d[key] = _walk(d[key], schema.get(ekey, {}))
+            if ekey != key:
+                d[ekey] = d[key]
+                del d[key]
+        return d
+
+    _walk(content, schema or {})
+
+    return ONDict(content)
+
+
+def _loadobj(field: Union[fields.Field, Any], value: Any) -> Any:
+    if isinstance(field, (fields.Field)):
+        value = field.from_obj(value)
+    return value

src/resconfig/io/paths.py

@@ -2,6 +2,7 @@
 from pathlib import Path
 
 from ..typing import FilePath
+from ..typing import Optional
 from . import ini
 from . import json
 from . import toml
@@ -15,13 +16,24 @@ class ConfigPath(type(Path())):
 
     module = None
 
-    def dump(self, content):
+    def dump(self, conf: "ONDict", schema: Optional["ONDict"] = None):
+        """Dump config to file at path.
+
+        Args:
+            conf: Configuration to dump.
+            schema: Configuration schema.
+        """
         with open(self, "w") as f:
-            self.module.dump(content, f)
+            self.module.dump(conf, f, schema)
+
+    def load(self, schema: Optional["ONDict"] = None) -> "ONDict":
+        """Load config from file at path.
 
-    def load(self):
+        Args:
+            schema: Configuration schema.
+        """
         with open(self) as f:
-            return self.module.load(f)
+            return self.module.load(f, schema)
 
     @classmethod
     def from_extension(cls, filename: FilePath) -> "ConfigPath":