From ff7d0e284cc39567c5d937a5428d53c881fd3594 Mon Sep 17 00:00:00 2001 From: Krzysztof Magusiak Date: Sun, 26 Nov 2023 19:02:24 +0100 Subject: [PATCH] pydantic draft --- alphaconf/__init__.py | 3 +- alphaconf/internal/configuration.py | 121 +++++++++++++++++++++------ alphaconf/internal/type_resolvers.py | 9 +- demo.ipynb | 16 ++-- example-simple.py | 29 ++++--- example-typed.py | 26 ++++-- 6 files changed, 142 insertions(+), 62 deletions(-) diff --git a/alphaconf/__init__.py b/alphaconf/__init__.py index ee9594e..2b7dad9 100644 --- a/alphaconf/__init__.py +++ b/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"] diff --git a/alphaconf/internal/configuration.py b/alphaconf/internal/configuration.py index 0443924..3e91f18 100644 --- a/alphaconf/internal/configuration.py +++ b/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 diff --git a/alphaconf/internal/type_resolvers.py b/alphaconf/internal/type_resolvers.py index f85a2f9..dc7494d 100644 --- a/alphaconf/internal/type_resolvers.py +++ b/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) diff --git a/demo.ipynb b/demo.ipynb index 7f4e069..4f9aa1c 100644 --- a/demo.ipynb +++ b/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" ] } diff --git a/example-simple.py b/example-simple.py index b47d787..7af35de 100755 --- a/example-simple.py +++ b/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: diff --git a/example-typed.py b/example-typed.py index c1e1bde..ece0717 100755 --- a/example-typed.py +++ b/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__':