pydantic draft

alphaconf/__init__.py

@@ -77,7 +77,7 @@ def run(
     app: Optional[Application] = None,
     **config,
 ) -> Optional[T]:
-    """Run this application
+    """Run this application (deprecated)
 
     If an application is not given, a new one will be created with configuration properties
     taken from the config. Also, by default logging is set up.
@@ -156,3 +156,4 @@ def __alpha_configuration():
 
 # Initialize configuration
 __alpha_configuration()
+__all__ = ["get", "setup_configuration", "set_application", "Application", "frozendict"]

alphaconf/internal/configuration.py

@@ -1,5 +1,7 @@
 import copy
 import os
+import typing
+import warnings
 from enum import Enum
 from typing import (
     Any,
@@ -16,7 +18,7 @@
 
 from omegaconf import Container, DictConfig, OmegaConf
 
-from .type_resolvers import convert_to_type
+from .type_resolvers import convert_to_type, pydantic
 
 T = TypeVar('T')
 
@@ -31,7 +33,7 @@ class RaiseOnMissingType(Enum):
 
 class Configuration:
     c: DictConfig
-    __type_path: MutableMapping[Type, str]
+    __type_path: MutableMapping[Type, Optional[str]]
     __type_value: MutableMapping[Type, Any]
     helpers: Dict[str, str]
 
@@ -96,7 +98,7 @@ def get(self, key: Union[str, Type], type=None, *, default=raise_on_missing):
             return value
         if isinstance(value, Container):
             value = OmegaConf.to_object(value)
-        if type is not None:
+        if type is not None and default is not None:
             value = convert_to_type(value, type)
         return value
 
@@ -124,37 +126,45 @@ def _merge(self, configs: Iterable[DictConfig]):
 
     def setup_configuration(
         self,
-        conf: Union[DictConfig, str, Dict],  # XXX Type[BaseModel]
-        helpers: Dict[str, str] = {},  # XXX deprecated arg?
+        conf: Union[DictConfig, dict, Any],
+        helpers: Dict[str, str] = {},
+        *,
+        path: str = "",
     ):
         """Add a default configuration
 
         :param conf: The configuration to merge into the global configuration
         :param helpers: Description of parameters used in argument parser helpers
+        :param path: The path to add the configuration to
         """
-        # merge the configurations
-        # TODO prepare_config in DictConfig?
-        # TODO type in values
+        if isinstance(conf, type):
+            # if already registered, set path to None
+            self.__type_path[conf] = None if conf in self.__type_path else path
+        if path and not path.endswith('.'):
+            path += "."
         if isinstance(conf, str):
+            warnings.warn("provide a dict directly", DeprecationWarning)
             created_config = OmegaConf.create(conf)
             if not isinstance(created_config, DictConfig):
                 raise ValueError("The config is not a dict")
             conf = created_config
         if isinstance(conf, DictConfig):
-            config = conf
+            config = self.__prepare_dictconfig(conf, path=path)
         else:
-            created_config = OmegaConf.create(Configuration._prepare_config(conf))
-            if not (created_config and isinstance(created_config, DictConfig)):
-                raise ValueError('Expecting a non-empty dict configuration')
+            created_config = self.__prepare_config(conf, path=path)
+            if not isinstance(created_config, DictConfig):
+                raise ValueError("Failed to convert to a DictConfig")
             config = created_config
+        # add path and merge
+        if path:
+            config = self.__add_path(config, path.rstrip("."))
         self._merge([config])
-        # setup helpers
-        for h_key in helpers:
-            key = h_key.split('.', 1)[0]
-            if not config or key not in config:
-                raise ValueError('Invalid helper not in configuration [%s]' % key)
         self.helpers.update(**helpers)
 
+    def add_helper(self, key, description):
+        """Assign a helper description"""
+        self.helpers[key] = description
+
     def from_environ(self, prefixes: Iterable[str]) -> DictConfig:
         """Load environment variables into a dict configuration"""
         from yaml.error import YAMLError  # type: ignore
@@ -176,14 +186,71 @@ def from_environ(self, prefixes: Iterable[str]) -> DictConfig:
                 OmegaConf.update(conf, name, value)
         return conf
 
-    @staticmethod
-    def _prepare_config(conf):
-        if not isinstance(conf, dict):
-            return conf
-        for k, v in conf.items():
+    def __prepare_dictconfig(
+        self, obj: DictConfig, path: str, recursive: bool = True
+    ) -> DictConfig:
+        sub_configs = []
+        for k, v in obj.items_ex(resolve=False):
+            if not isinstance(k, str):
+                raise ValueError("Expecting only str instances in dict")
+            if recursive:
+                v = self.__prepare_config(v, path + k + ".")
             if '.' in k:
-                parts = k.split('.')
-                k = parts[0]
-                v = {'.'.join(parts[1:]): v}
-            conf[k] = Configuration._prepare_config(v)
-        return conf
+                obj.pop(k)
+                sub_configs.append(self.__add_path(v, k))
+        if sub_configs:
+            obj = cast(DictConfig, OmegaConf.unsafe_merge(obj, *sub_configs))
+        return obj
+
+    def __prepare_config(self, obj, path):
+        if isinstance(obj, DictConfig):
+            return self.__prepare_dictconfig(obj, path)
+        if pydantic:
+            obj = self.__prepare_pydantic(obj, path)
+        if isinstance(obj, dict):
+            result = {}
+            changed = False
+            for k, v in obj.items():
+                result[k] = nv = self.__prepare_config(v, path + k + ".")
+                changed |= v is not nv
+            if not changed:
+                result = obj
+            return self.__prepare_dictconfig(OmegaConf.create(result), path, recursive=False)
+        return obj
+
+    def __prepare_pydantic(self, obj, path):
+        if isinstance(obj, pydantic.BaseModel):
+            # pydantic instance, prepare helpers
+            self.__prepare_pydantic(type(obj), path)
+            return obj.model_dump(mode="json")
+        # parse typing recursively for documentation
+        for t in typing.get_args(obj):
+            self.__prepare_pydantic(t, path)
+        # check if not a type
+        if not isinstance(obj, type):
+            return obj
+        # prepare documentation from types
+        if issubclass(obj, pydantic.BaseModel):
+            # pydantic type
+            defaults = {}
+            for k, field in obj.model_fields.items():
+                check_type = True
+                if field.default is not pydantic.fields._Unset:
+                    defaults[k] = field.default
+                    check_type = not bool(defaults[k])
+                elif field.is_required():
+                    defaults[k] = "???"
+                else:
+                    defaults[k] = None
+                if desc := (field.description or field.title):
+                    self.add_helper(path + k, desc)
+                if check_type and field.annotation:
+                    self.__prepare_pydantic(field.annotation, path + k + ".")
+            return defaults
+        return None
+
+    @staticmethod
+    def __add_path(config: Any, path: str) -> DictConfig:
+        for part in reversed(path.split(".")):
+            config = OmegaConf.create({part: config})
+        return config

alphaconf/internal/type_resolvers.py

@@ -38,6 +38,7 @@ def parse_bool(value) -> bool:
     'read_strip': lambda s: read_text(s).strip(),
     'read_bytes': lambda s: Path(s).expanduser().read_bytes(),
 }
+_type = type
 
 # register resolved from strings
 for _name, _function in TYPE_CONVERTER.items():
@@ -52,7 +53,11 @@ def convert_to_type(value, type):
     :param type: A class or a callable used to convert the value
     :return: Result of the callable
     """
-    if pydantic and isinstance(type, pydantic.BaseModel):
-        return type.model_validate(value)
+    if pydantic:
+        if issubclass(type, pydantic.BaseModel):
+            type.model_construct
+            return type.model_validate(value)
+        if isinstance(type, _type):
+            return pydantic.TypeAdapter(type).validate_python(value)
     type = TYPE_CONVERTER.get(type, type)
     return type(value)

demo.ipynb

@@ -34,7 +34,6 @@
       "\n",
       "positional arguments:\n",
       "  key=value                   Configuration items\n",
-      "  server                      Arguments for the demo\n",
       "  show                        The name of the selection to show\n",
       "  exception                   If set, raise an exception\n",
       "\u001b[0m"
@@ -82,12 +81,10 @@
       "  logging:\n",
       "  - default\n",
       "  - none\n",
+      "exception: false\n",
       "server:\n",
-      "  url: http://default\n",
+      "  name: test_server\n",
       "  user: ${oc.env:USER}\n",
-      "  home: '~'\n",
-      "show: false\n",
-      "exception: false\n",
       "application:\n",
       "  name: example\n",
       "  version: '0.1'\n",
@@ -113,11 +110,10 @@
      "output_type": "stream",
      "text": [
       "app: example\n",
+      "server.name test_server\n",
       "server.user: k\n",
-      "server.home /home/k\n",
       "INFO:root:['init'] The app is running...\n",
       "INFO:root:Just a log\n",
-      "INFO:root:['finished'] Application end.\n",
       "\u001b[0m"
      ]
     }
@@ -136,11 +132,11 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "INFO \u001b[32mApplication start (example-inv: InvokeApplication.run_program)\u001b[0m\n",
+      "INFO \u001b[32mStart (example-inv: InvokeApplication.run_program)\u001b[0m\n",
       "INFO \u001b[32mHello\u001b[0m\n",
       "INFO \u001b[32mBackup: me\u001b[0m\n",
-      "INFO \u001b[32mParam: [4] and in alphaconf [4]\u001b[0m\n",
-      "INFO \u001b[32mApplication end.\u001b[0m\n",
+      "INFO \u001b[32mParam: [4]\u001b[0m\n",
+      "INFO \u001b[32mEnd.\u001b[0m\n",
       "\u001b[0m"
      ]
     }

example-simple.py

@@ -1,6 +1,5 @@
 #!/usr/bin/env python3
 import logging
-from pathlib import Path
 from typing import Optional
 
 from pydantic import BaseModel, Field
@@ -9,32 +8,32 @@
 import alphaconf.logging_util
 
 
-class Conn(BaseModel):
-    url: str
-    user: str = ""
-    home: Path = Path("~")
-
-
 class Opts(BaseModel):
-    server: Conn = Field(Conn(url="http://default"), description="Arguments for the demo")
     show: Optional[str] = Field(None, description="The name of the selection to show")
     exception: bool = Field(False, description="If set, raise an exception")
 
 
 # adding a default configuration
 # these will be merged with the application
-alphaconf.setup_configuration({"": Opts})
-alphaconf.setup_configuration({"server.user": "${oc.env:USER}"})
+alphaconf.setup_configuration(Opts)
+alphaconf.setup_configuration(
+    {
+        "server": {
+            "name": "test_server",
+            "user": "${oc.env:USER}",
+        }
+    }
+)
 
 
 def main():
     """Simple demo of alphaconf"""
 
     # get the application name from the configuration
-    print('app:', alphaconf.global_configuration.c.application.name)
+    print('app:', alphaconf.get("application.name"))
     # shortcut version to get a configuration value
+    print('server.name', alphaconf.get('server.name'))
     print('server.user:', alphaconf.get('server.user'))
-    print('server.home', alphaconf.get('server.home', Path))
 
     # you can set additional dynamic values in the logging
     context_value = ['init']
@@ -46,11 +45,11 @@ def main():
 
     logging.info('Just a log')
     # show configuration
-    value = alphaconf.get('show', str)
-    if value and (value := alphaconf.get(value)):
+    value = alphaconf.get('show', str, default=None)
+    if value and (value := alphaconf.get(value, default=None)):
         print(value)
     # log an exception if we have it in the configuration
-    if alphaconf.get('exception'):
+    if alphaconf.get('exception', default=False):
         try:
             raise RuntimeError("Asked to raise something")
         except Exception:

example-typed.py

@@ -1,27 +1,39 @@
 #!/usr/bin/env python3
 import logging
 from datetime import date
+from pathlib import Path
 from typing import Optional
 
-import pydantic
+from pydantic import BaseModel, Field
 
 import alphaconf.cli
 
 
-class MyConfiguration(pydantic.BaseModel):
-    name: Optional[str] = None
-    """name variable"""
-    dd: Optional[date] = None
+class Conn(BaseModel):
+    url: str = Field("http://github.com", title="Some URL")
+    home: Path = Path("~")
 
 
-alphaconf.setup_configuration({"c": MyConfiguration})
+class MyConfiguration(BaseModel):
+    name: Optional[str] = Field(None, title="Some name to show")
+    some_date: Optional[date] = None
+    connection: Optional[Conn] = None
+
+
+alphaconf.setup_configuration(MyConfiguration, path="c")
 
 
 def main():
     """Typed configuration example"""
     logging.info('Got configuration name: %s', alphaconf.get('c.name'))
     c = alphaconf.get(MyConfiguration)
-    logging.info("Found configuration object:", c)
+    logging.info("Found configuration object: %s", c)
+    if c.connection:
+        logging.info(
+            'connection.home: %s (%s)',
+            alphaconf.get('c.connection.home', default='unset'),
+            c.connection.home,
+        )
 
 
 if __name__ == '__main__':