From 6a78363d5a9b82b0ac9042687a7d4a8209c9325e Mon Sep 17 00:00:00 2001 From: Adrian Cederberg Date: Tue, 16 Jul 2024 08:11:15 -0600 Subject: [PATCH] feature(loader): Removed ``__yaml_files__`` and ``__yaml_reload__``. Removed tests concerning their use. Moved ``CreateYamlSettings`` to ``settings.py`` and validators to ``loader.py``. Added ``util.py``. --- tests/test_basic.py | 92 +++++------ yaml_settings_pydantic/__init__.py | 235 +---------------------------- yaml_settings_pydantic/loader.py | 84 ++++++++++- yaml_settings_pydantic/settings.py | 107 +++++++++++++ yaml_settings_pydantic/util.py | 12 ++ 5 files changed, 255 insertions(+), 275 deletions(-) create mode 100644 yaml_settings_pydantic/settings.py create mode 100644 yaml_settings_pydantic/util.py diff --git a/tests/test_basic.py b/tests/test_basic.py index 8e51096..8b70184 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -35,8 +35,10 @@ def create_settings(reload: Any | None = None, files: Any | None = None) -> Any: "Settings", (BaseYamlSettings,), { - "__yaml_reload__": reload or False, - "__yaml_files__": files or set(file_dummies), + "model_config": YamlSettingsConfigDict( + yaml_reload=reload if reload is not None else False, + yaml_files=files or set(file_dummies), + ) }, ) @@ -44,24 +46,24 @@ def create_settings(reload: Any | None = None, files: Any | None = None) -> Any: yaml_settings = CreateYamlSettings(Settings) yaml_settings() - assert not yaml_settings.reload, "Should not reload." + assert not yaml_settings.settings_cls.model_config.get( + "yaml_reload" + ), "Should not reload." # Malform a file. - bad: Path = Settings.__yaml_files__.pop() + bad: Path = Settings.model_config["yaml_files"].pop() with bad.open("w") as file: yaml.dump([], file) - # # NOTE: Loading should not be an error as the files should not be reloaded. + # NOTE: Loading should not be an error as the files should not be reloaded. yaml_settings() - # - # # NOTE: Test reloading with bad file. - # # This could be called without the args as mutation is visible - # # to fn. - Settings = create_settings(reload=False) - yaml_settings = CreateYamlSettings(Settings) + # NOTE: Test reloading with bad file. + # This could be called without the args as mutation is visible + # to fn. + Settings = create_settings(reload=False) with pytest.raises(ValueError) as err: - yaml_settings() + yaml_settings = CreateYamlSettings(Settings) assert bad.as_posix() in str(err.value), "Missing required path in message." @@ -71,37 +73,37 @@ def create_settings(reload: Any | None = None, files: Any | None = None) -> Any: yaml_settings() def from_model_config( - self, **kwargs: Any + self, *, load: bool = False, **kwargs: Any ) -> tuple[CreateYamlSettings, type[BaseYamlSettings]]: Settings = type( "Settings", (BaseYamlSettings,), {"model_config": YamlSettingsConfigDict(**kwargs)}, # type: ignore ) - return CreateYamlSettings(Settings), Settings - - def test_dunders_have_priority(self) -> None: - init_reload = True - foo_bar: Path = Path("foo-bar.yaml") - yaml_settings, Settings = self.from_model_config( - yaml_files={foo_bar}, - yaml_reload=init_reload, - ) - - default = DEFAULT_YAML_FILE_CONFIG_DICT - assert yaml_settings.files == {foo_bar: default} - assert yaml_settings.reload == init_reload - - final_files: set[Path] = {Path("spam-eggs.yaml")} - OverwriteSettings = type( - "OverwriteSettings", - (Settings,), - {"__yaml_files__": final_files}, - ) - yaml_settings = CreateYamlSettings(OverwriteSettings) - - assert yaml_settings.files == {Path("spam-eggs.yaml"): default} - assert yaml_settings.reload == init_reload + return CreateYamlSettings(Settings, load=load), Settings + + # def test_dunders_have_priority(self) -> None: + # init_reload = True + # foo_bar: Path = Path("foo-bar.yaml") + # yaml_settings, Settings = self.from_model_config( + # yaml_files={foo_bar}, + # yaml_reload=init_reload, + # ) + # + # default = DEFAULT_YAML_FILE_CONFIG_DICT + # assert yaml_settings.files == {foo_bar: default} + # assert yaml_settings.reload == init_reload + # + # final_files: set[Path] = {Path("spam-eggs.yaml")} + # OverwriteSettings = type( + # "OverwriteSettings", + # (Settings,), + # {"__yaml_files__": final_files}, + # ) + # yaml_settings = CreateYamlSettings(OverwriteSettings) + # + # assert yaml_settings.files == {Path("spam-eggs.yaml"): default} + # assert yaml_settings.reload == init_reload @pytest.mark.parametrize( "yaml_files", @@ -114,10 +116,12 @@ def test_dunders_have_priority(self) -> None: def test_hydration_yaml_files(self, yaml_files: Any) -> None: make, _ = self.from_model_config(yaml_files=yaml_files) - assert len(make.files) == 1, "Exactly one file." - assert isinstance(make.files, dict), "Files should hydrate to a ``dict``." + assert len(make.yaml_files_configs) == 1, "Exactly one file." + assert isinstance( + make.yaml_files_configs, dict + ), "Files should hydrate to a ``dict``." assert ( - foo := make.files.get(Path("foo.yaml")) + foo := make.yaml_files_configs.get(Path("foo.yaml")) ), "An entry should exist under key ``Path('foo.yaml')``." assert isinstance(foo, dict), "`foo` must be a dictionary." assert foo.get("required"), "Required is always ``True`` by default." @@ -133,7 +137,7 @@ def test_yaml_not_required(self) -> None: ), }, ) - if not make.files.get(Path("foo.yaml")): + if not make.yaml_files_configs.get(Path("foo.yaml")): raise ValueError make.load() @@ -161,7 +165,7 @@ def test_envvar(self, tmp_path: pathlib.Path) -> None: ) } ) - assert set(make.files.keys()) == {path_default} + assert set(make.yaml_files_configs.keys()) == {path_default} class SettingsModel(BaseModel): whatever: Annotated[str, Field(default="whatever")] @@ -183,7 +187,7 @@ class Settings(SettingsModel, _Settings): ... # type: ignore with mock.patch.dict(os.environ, {"FOO_PATH": str(path_other)}, clear=True): filepath_resolved = resolve_filepaths( path_default, - make.files[path_default], + make.yaml_files_configs[path_default], ) assert filepath_resolved == path_other @@ -194,7 +198,7 @@ class Settings(SettingsModel, _Settings): ... # type: ignore with mock.patch.dict(os.environ, {"FOO_PATH": ""}, clear=True): filepath_resolved = resolve_filepaths( path_default, - make.files[path_default], + make.yaml_files_configs[path_default], ) assert filepath_resolved == path_default diff --git a/yaml_settings_pydantic/__init__.py b/yaml_settings_pydantic/__init__.py index e59ce87..3cede6e 100644 --- a/yaml_settings_pydantic/__init__.py +++ b/yaml_settings_pydantic/__init__.py @@ -15,239 +15,21 @@ from __future__ import annotations -import logging -from collections.abc import Sequence -from os import environ -from pathlib import Path, PosixPath -from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Literal, TypeVar +from typing import TYPE_CHECKING, ClassVar -from pydantic.fields import FieldInfo from pydantic_settings import BaseSettings, PydanticBaseSettingsSource -from typing_extensions import Doc -from yaml_settings_pydantic import loader +from yaml_settings_pydantic import loader, util from yaml_settings_pydantic.loader import ( DEFAULT_YAML_FILE_CONFIG_DICT, + YamlFileConfigDict, YamlSettingsConfigDict, + resolve_filepaths, ) +from yaml_settings_pydantic.settings import CreateYamlSettings __version__ = "2.3.1" -logger = logging.getLogger("yaml_settings_pydantic") -if environ.get("YAML_SETTINGS_PYDANTIC_LOGGER") == "true": - logging.basicConfig(level=logging.DEBUG) - logger.setLevel(logging.DEBUG) -T = TypeVar("T") - - -class CreateYamlSettings(PydanticBaseSettingsSource): - """Create a ``yaml`` setting loader middleware. - - - Note that the following fields can be set using dunder ``ClassVars`` or - ``model_config`` on ``settings_cls.model_config``. - """ - - # Info - - files: Annotated[ - dict[Path, loader.YamlFileConfigDict], - Doc( - "``YAML`` or ``JSON`` files to load and loading specifications (" - "in the form of :class:`YamlFileConfigDict`)." - ), - ] - reload: Annotated[ - bool, - Doc( - "When ``True```, reload files specified in :param:`files` when a " - "new instance is created. Default is `False`." - ), - ] - - # State - _loaded: Annotated[ - dict[str, Any] | None, - Doc("Loaded file(s) content."), - ] - - # ----------------------------------------------------------------------- # - # Top level stuff. - - def __init__(self, settings_cls: type[BaseSettings]): - self.reload = self.validate_reload(settings_cls) - self.files = self.validate_files(settings_cls) - self._loaded = None - - def __call__(self) -> dict[str, Any]: - """Yaml settings loader for a single file. - - :returns: Yaml from :attr:`files` unmarshalled and combined by update. - """ - - return self.loaded - - @property - def loaded(self) -> dict[str, Any]: - """Loaded file(s) content. - - Always loads content the first time. On subsequent calls, returns - will return previously loaded content if :attr:`reload` is `False`, - otherwise returns output by calling :meth:`load`. - """ - if self.reload: - logger.debug("Reloading configuration files.") - self._loaded = self.load() - elif self._loaded is None: - logger.debug("Loading configuration files. Should not reload.") - self._loaded = self.load() - - return self._loaded - - def get_field_value( - self, - field: FieldInfo, - field_name: str, - ) -> tuple[Any, str, bool]: - """Required by pydantic.""" - - v = self.loaded.get(field_name) - return (v, field_name, False) - - # ----------------------------------------------------------------------- # - # Field validation. - - def validate_reload(self, settings_cls: type[BaseSettings]) -> bool: - logger.debug("`%s` validating `%s`.", self, settings_cls.__name__) - reload: bool = self.get_settings_cls_value( - settings_cls, - "reload", - True, - ) - - return reload - - def validate_files( - self, settings_cls: type[BaseSettings] - ) -> dict[Path, loader.YamlFileConfigDict]: - """Validate ``model_config["files"]``.""" - - found_value: dict[Path, loader.YamlFileConfigDict] | str | Sequence[str] | None - found_value = self.get_settings_cls_value(settings_cls, "files", None) - item = f"{settings_cls.__name__}.model_config.yaml_files" - - # NOTE: Validate is dict/set/Path/str - if found_value is None: - raise ValueError(f"`{item}` cannot be `None`.") - elif ( - not isinstance(found_value, Path) - and not isinstance(found_value, str) - and not isinstance(found_value, set) - and not isinstance(found_value, dict) - ): - msg = "`{0}` must be a sequence or set, got type `{1}`." - raise ValueError(msg.format(item, type(found_value))) - # NOTE: Not including makes the editor think the the code below is - # unreachable, I do not know why, so the ``else`` statement shall - # remain. - else: - ... - - # NOTE: If its a string/``Path``, make it into a tuple. If it is anything - # else just leave it. - values: ( - tuple[Path, ...] - | dict[str, loader.YamlFileConfigDict] - | dict[Path, loader.YamlFileConfigDict] - ) - if isinstance(found_value, PosixPath): - logger.debug(f"`{item}` was a PosixPath.") - values = (found_value,) - elif isinstance(found_value, str): - logger.debug(f"`{item}` was a String.") - values = (Path(found_value),) - else: - values = found_value - - keys_invalid = {item for item in values if not isinstance(item, Path)} - if len(keys_invalid): - raise ValueError( - "All items in `files` must be strings. The following are " - f"not strings: `{keys_invalid}`." - ) - - # NOTE: Create dictionary if the sequence is not a dictionary. - files: dict[Path, loader.YamlFileConfigDict] - if not isinstance(values, dict): - files = { - ( - k if isinstance(k, Path) else Path(k) - ): loader.DEFAULT_YAML_FILE_CONFIG_DICT.copy() - for k in values - } - elif any(not isinstance(v, dict) for v in values.values()): - raise ValueError(f"`{item}` values must have type `dict`.") - elif not len(values): - raise ValueError("`files` cannot have length `0`.") - else: - for k, v in values.items(): - vv = loader.DEFAULT_YAML_FILE_CONFIG_DICT.copy() - vv.update(v) - values[k] = v - files = values - - return files - - def get_settings_cls_value( - self, - settings_cls: Any, - field: Literal["files", "reload"], - default: T, - ) -> T: - """Look for and return an attribute :param:`field` on - :param:`settings_cls` and then :attr:`settings_cls.model_config`, if - neither of these are found return :param:`default`. - """ - # Bc logging - _msg = "Looking for field `%s` as `%s` on `%s`." - _msg_found = _msg.replace("Looking for", "Found") - - # Bc naming - cls_field = f"__yaml_{field}__" - config_field = f"yaml_{field}" - - # Look for dunder source - logger.debug(_msg, f"__{field}__", config_field, "settings_cls") - out = default - if (dunder := getattr(settings_cls, cls_field, None)) is not None: - logger.debug(_msg_found, field, config_field, "settings_cls") - return dunder - - # Look for config source - logger.debug(_msg, field, config_field, "settings_cls.model_config") - from_conf = settings_cls.model_config.get(config_field) - if from_conf is not None: - logger.debug( - _msg_found, - field, - config_field, - "settings_cls.model_config", - ) - return from_conf - - # Return defult - logger.debug("Using default `%s` for field `%s`.", default, field) - return out - - # ----------------------------------------------------------------------- # - # Loading - - def load(self) -> dict[str, Any]: - """Load data and validate that it is sufficiently shaped for - ``BaseSettings``. - """ - - self._yaml_data = (yaml_data := loader.load_yaml_data(self.files)) - return loader.validate_yaml_data(yaml_data) +logger = util.get_logger(__name__) class BaseYamlSettings(BaseSettings): @@ -270,9 +52,6 @@ class BaseYamlSettings(BaseSettings): # thinks that `model_config` is a model field name. model_config: ClassVar[loader.YamlSettingsConfigDict] - __yaml_files__: ClassVar[Sequence[str] | None] - __yaml_reload__: ClassVar[bool | None] - @classmethod def settings_customise_sources( cls, @@ -299,8 +78,6 @@ def settings_customise_sources( ) -from yaml_settings_pydantic.loader import YamlFileConfigDict, resolve_filepaths - __all__ = ( "resolve_filepaths", "CreateYamlSettings", diff --git a/yaml_settings_pydantic/loader.py b/yaml_settings_pydantic/loader.py index 34ead8b..d6f6210 100644 --- a/yaml_settings_pydantic/loader.py +++ b/yaml_settings_pydantic/loader.py @@ -2,8 +2,8 @@ from collections.abc import Sequence from os import environ -from pathlib import Path -from typing import Annotated, Any, NotRequired, TypedDict +from pathlib import Path, PosixPath +from typing import Annotated, Any, NotRequired, Set, TypedDict import jsonpath_ng import yaml @@ -101,6 +101,86 @@ def resolve_filepaths(fp: Path, fp_config: YamlFileConfigDict) -> Path: return fp_final +def validate_yaml_settings_config_reload( + yaml_settings_config: YamlSettingsConfigDict, +) -> bool: + from_conf = yaml_settings_config.get("yaml_reload") + return from_conf if from_conf is not None else True + + +def validate_yaml_settings_config_files( + yaml_settings_config: YamlSettingsConfigDict, +) -> dict[Path, YamlFileConfigDict]: + """Validate ``yaml_settings_config['files']``.""" + + found_value = yaml_settings_config.get("yaml_files") + item = "model_config.yaml_files" + + # NOTE: Validate is dict/set/Path/str + if found_value is None: + raise ValueError(f"`{item}` cannot be `None`.") + elif ( + not isinstance(found_value, Path) + and not isinstance(found_value, str) + and not isinstance(found_value, set) + and not isinstance(found_value, dict) + ): + msg = "`{0}` must be a sequence or set, got type `{1}`." + raise ValueError(msg.format(item, type(found_value))) + # NOTE: Not including makes the editor think the the code below is + # unreachable, I do not know why, so the ``else`` statement shall + # remain. + else: + ... + + # NOTE: If its a string/``Path``, make it into a tuple. If it is anything + # else just leave it. + values: ( + tuple[Path, ...] + | dict[str, YamlFileConfigDict] + | dict[Path, YamlFileConfigDict] + | set[Path] + | set[str] + ) + if isinstance(found_value, Path): + # logger.debug(f"`{item}` was a PosixPath.") + values = (found_value,) + elif isinstance(found_value, str): + # logger.debug(f"`{item}` was a String.") + values = (Path(found_value),) + else: + values = found_value + + keys_invalid = {item for item in values if not isinstance(item, Path)} + if len(keys_invalid): + raise ValueError( + "All items in `files` must be strings. The following are " + f"not strings: `{keys_invalid}`." + ) + + # NOTE: Create dictionary if the sequence is not a dictionary. + files: dict[Path, YamlFileConfigDict] + if not isinstance(values, dict): + files = { + ( + k if isinstance(k, Path) else Path(k) + ): DEFAULT_YAML_FILE_CONFIG_DICT.copy() + for k in values + } + elif any(not isinstance(v, dict) for v in values.values()): + raise ValueError(f"`{item}` values must have type `dict`.") + elif not len(values): + raise ValueError("`files` cannot have length `0`.") + else: + files = {} + for k, v in values.items(): + vv = DEFAULT_YAML_FILE_CONFIG_DICT.copy() + vv.update(v) + files[k if isinstance(k, Path) else Path(k)] = v + + return files + + def load_yaml_data( yaml_file_configs: dict[Path, YamlFileConfigDict] ) -> dict[Path, YamlFileData]: diff --git a/yaml_settings_pydantic/settings.py b/yaml_settings_pydantic/settings.py new file mode 100644 index 0000000..70f106f --- /dev/null +++ b/yaml_settings_pydantic/settings.py @@ -0,0 +1,107 @@ +from pathlib import Path +from typing import Annotated, Any + +from pydantic.fields import FieldInfo +from pydantic_settings import BaseSettings, PydanticBaseSettingsSource +from typing_extensions import Doc + +from yaml_settings_pydantic import loader, util + +logger = util.get_logger(__name__) + + +class CreateYamlSettings(PydanticBaseSettingsSource): + """Create a ``yaml`` setting loader middleware. + + + Note that the following fields can be set using dunder ``ClassVars`` or + ``model_config`` on ``settings_cls.model_config``. + """ + + yaml_reload: Annotated[ + bool, + Doc( + "When ``True```, reload files specified in :param:`files` when a " + "new instance is created. Default is `False`." + ), + ] + yaml_files_configs: Annotated[ + dict[Path, loader.YamlFileConfigDict], + Doc( + "``YAML`` or ``JSON`` files to load and loading specifications (" + "in the form of :class:`YamlFileConfigDict`)." + ), + ] + yaml_files_data: Annotated[ + dict[Path, loader.YamlFileData], + Doc("Hydrated data of ``yaml_file_configs``."), + ] + + _loaded: Annotated[ + dict[str, Any] | None, + Doc("Loaded file(s) content."), + ] + + # ----------------------------------------------------------------------- # + # Top level stuff. + + def __init__(self, settings_cls: type[BaseSettings], *, load: bool = True): + self.settings_cls = settings_cls + + conf: loader.YamlSettingsConfigDict + conf = settings_cls.model_config # type: ignore + + yaml_files = loader.validate_yaml_settings_config_files(conf) + yaml_reload = loader.validate_yaml_settings_config_reload(conf) + + # if (yaml_files_configs := settings_cls.model_config.get("yaml_files")) is None: + # raise ValueError("Missing ``model_config['yaml_files_configs']``.") + + self.yaml_files_configs = yaml_files + self.yaml_reload = yaml_reload + self._loaded = None + if load: + self.load() + + def __call__(self) -> dict[str, Any]: + """Yaml settings loader for a single file. + + :returns: Yaml from :attr:`files` unmarshalled and combined by update. + """ + + return self.loaded + + @property + def loaded(self) -> dict[str, Any]: + """Loaded file(s) content. + + Always loads content the first time. On subsequent calls, returns + will return previously loaded content if :attr:`reload` is `False`, + otherwise returns output by calling :meth:`load`. + """ + if self.yaml_reload: + logger.debug("Reloading configuration files.") + self._loaded = self.load() + elif self._loaded is None: + logger.debug("Loading configuration files. Should not reload.") + self._loaded = self.load() + + return self._loaded + + def load(self) -> dict[str, Any]: + """Load data and validate that it is sufficiently shaped for + ``BaseSettings``. + """ + + self.yaml_files_data = loader.load_yaml_data(self.yaml_files_configs) + return loader.validate_yaml_data(self.yaml_files_data) + + def get_field_value( + self, + field: FieldInfo, + field_name: str, + ) -> tuple[Any, str, bool]: + """Required by pydantic.""" + + v = self.loaded.get(field_name) + return (v, field_name, False) diff --git a/yaml_settings_pydantic/util.py b/yaml_settings_pydantic/util.py new file mode 100644 index 0000000..d9697f9 --- /dev/null +++ b/yaml_settings_pydantic/util.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +import logging +from os import environ + + +def get_logger(name): + logger = logging.getLogger(name) + if environ.get("YAML_SETTINGS_PYDANTIC_LOGGER") == "true": + logging.basicConfig(level=logging.DEBUG) + logger.setLevel(logging.DEBUG) + return logger