Adding support for pydantic-settings > 2.2.0

pyproject.toml

@@ -34,12 +34,10 @@ dependencies = [
     "jinja2",
     "logzero",
     "orjson",
+    "packaging",
     "pydantic[email]",
-    "pydantic-settings<2.2.2",
-    # This is not an ideal pin, but diue to the issue highlighted in https://github.com/pydantic/pydantic-settings/issues/245
-    # and the non-semver compliant versioning in pydantic-settings, we need to add this pin
-    # this version would change the API behaviour for nskit as it will also ignore additional
-    # inputs in the python initialisation so We will pin to version < 2.2.0
+    "pydantic-settings>=2.2.0, <2.3.0",
+    # The issue highlighted in https://github.com/pydantic/pydantic-settings/issues/245 has been mitigated by a wrapper and additional parameter for dotenv_allow in the model config
     "python-dotenv",
     "ruamel.yaml",
     "tomlkit",
@@ -227,14 +225,15 @@ import-order-style="appnexus"
 
 [tool.isort]
 profile = "appnexus"
-src_paths = ['src/']
-known_first_party = [
+src_paths = ["src/"]
+known_first_party = ["nskit"
 ]
 
-known_application = 'nskit*'
+known_application = "nskit*"
 force_alphabetical_sort_within_sections = true
 force_sort_within_sections = true
 reverse_relative = true
+combine_as_imports = true
 
 [tool.pytest]
 python_classes = false

src/nskit/__init__.py

@@ -31,11 +31,15 @@ def __get_version() -> str:
         # Use the metadata
         import sys
         if sys.version_info.major >= 3 and sys.version_info.minor >= 8:
-            from importlib.metadata import PackageNotFoundError
-            from importlib.metadata import version as parse_version
+            from importlib.metadata import (
+                PackageNotFoundError,
+                version as parse_version,
+            )
         else:
-            from importlib_metadata import PackageNotFoundError
-            from importlib_metadata import version as parse_version  # type: ignore
+            from importlib_metadata import (  # type: ignore
+                PackageNotFoundError,
+                version as parse_version,
+            )
         try:
             version = parse_version("nskit")
         except PackageNotFoundError:

src/nskit/common/configuration/__init__.py

@@ -7,23 +7,39 @@
 """
 from __future__ import annotations
 
-from typing import Any
+from pathlib import Path
+from typing import Any, Optional
 
-from pydantic_settings import BaseSettings as _BaseSettings
-from pydantic_settings import PydanticBaseSettingsSource as _PydanticBaseSettingsSource
-from pydantic_settings import SettingsConfigDict as _SettingsConfigDict
+from pydantic.config import ExtraValues
+from pydantic_settings import (
+    BaseSettings as _BaseSettings,
+    PydanticBaseSettingsSource as _PydanticBaseSettingsSource,
+    SettingsConfigDict as _SettingsConfigDict,
+)
+from pydantic_settings.sources import PathType
 
 from nskit.common.configuration.mixins import PropertyDumpMixin
-from nskit.common.configuration.sources import FileConfigSettingsSource
+from nskit.common.configuration.sources import (
+    DotEnvSettingsSource,
+    JsonConfigSettingsSource,
+    TomlConfigSettingsSource,
+    YamlConfigSettingsSource,
+)
 from nskit.common.io import json, toml, yaml
 
 
+class SettingsConfigDict(_SettingsConfigDict):
+    """Customised Settings Config Dict."""
+    dotenv_extra: Optional[ExtraValues] = 'ignore'
+    config_file: Optional[PathType] = None
+    config_file_encoding: Optional[str] = None
+
+
 class BaseConfiguration(PropertyDumpMixin, _BaseSettings):
     """A Pydantic BaseSettings type object with Properties included in model dump, and yaml and toml integrations."""
 
-    model_config = _SettingsConfigDict(env_file_encoding='utf-8')
+    model_config = SettingsConfigDict(env_file_encoding='utf-8')
 
-    @classmethod
     def settings_customise_sources(
         cls,
         settings_cls: type[_BaseSettings],
@@ -33,13 +49,50 @@ def settings_customise_sources(
         file_secret_settings: _PydanticBaseSettingsSource,
     ) -> tuple[_PydanticBaseSettingsSource, ...]:
         """Create settings loading, including the FileConfigSettingsSource."""
-        # TODO This probably needs a tweak to handle complex structures.
+        config_files = cls.model_config.get('config_file')
+        config_file_encoding = cls.model_config.get('config_file_encoding')
+        file_types = {'json' : ['.json', '.jsn'],
+                      'yaml': ['.yaml', '.yml'],
+                      'toml': ['.toml', '.tml']}
+        if config_files:
+            if isinstance(config_files, (Path, str)):
+                config_files = [config_files]
+        else:
+            config_files = []
+
+        split_config_files = {}
+        for file_type, suffixes in file_types.items():
+            original = cls.model_config.get(f'{file_type}_file')
+            if original and isinstance(original, (Path, str)):
+                split_config_files[file_type] = [original]
+            elif original:
+                split_config_files[file_type] = original
+            else:
+                split_config_files[file_type] = []
+            for config_file in config_files:
+                if Path(config_file).suffix.lower() in suffixes:
+                    split_config_files[file_type].append(config_file)
         return (
-            FileConfigSettingsSource(settings_cls),
             init_settings,
             env_settings,
-            dotenv_settings,
-            file_secret_settings,
+            JsonConfigSettingsSource(settings_cls,
+                                     split_config_files['json'],
+                                     cls.model_config.get('json_file_encoding') or config_file_encoding),
+            YamlConfigSettingsSource(settings_cls,
+                                     split_config_files['yaml'],
+                                     cls.model_config.get('yaml_file_encoding') or config_file_encoding),
+            TomlConfigSettingsSource(settings_cls,
+                                     split_config_files['toml']),
+            DotEnvSettingsSource(settings_cls,
+                                 dotenv_settings.env_file,
+                                 dotenv_settings.env_file_encoding,
+                                 dotenv_settings.case_sensitive,
+                                 dotenv_settings.env_prefix,
+                                 dotenv_settings.env_nested_delimiter,
+                                 dotenv_settings.env_ignore_empty,
+                                 dotenv_settings.env_parse_none_str,
+                                 cls.model_config.get('dotenv_extra', 'ignore')),
+            file_secret_settings
         )
 
     def model_dump_toml(

src/nskit/common/configuration/sources.py

@@ -1,69 +1,107 @@
 """Add settings sources."""
+from __future__ import annotations as _annotations
+
 from pathlib import Path
-from typing import Any, Dict, Tuple
+from typing import Any
 
-from pydantic.fields import FieldInfo
-from pydantic_settings import PydanticBaseSettingsSource
+from pydantic.config import ExtraValues
+from pydantic_settings import BaseSettings
+from pydantic_settings.sources import (
+    DotEnvSettingsSource as _DotEnvSettingsSource,
+    DotenvType,
+    ENV_FILE_SENTINEL,
+    JsonConfigSettingsSource as _JsonConfigSettingsSource,
+    TomlConfigSettingsSource as _TomlConfigSettingsSource,
+    YamlConfigSettingsSource as _YamlConfigSettingsSource,
+)
 
 from nskit.common.io import json, toml, yaml
 
 
-class FileConfigSettingsSource(PydanticBaseSettingsSource):
-    """A simple settings source class that loads variables from a parsed file.
+class JsonConfigSettingsSource(_JsonConfigSettingsSource):
+    """Use the nskit.common.io.json loading to load settings from a json file."""
+    def _read_file(self, file_path: Path) -> dict[str, Any]:
+        encoding = self.json_file_encoding or 'utf-8'
+        file_contents = file_path.read_text(encoding)
+        return json.loads(file_contents)
+
+    def __call__(self):
+        """Make the file reading at the source instantiation."""
+        self.init_kwargs = self._read_files(self.json_file_path)
+        return super().__call__()
+
+
+class TomlConfigSettingsSource(_TomlConfigSettingsSource):
+    """Use the nskit.common.io.toml loading to load settings from a toml file."""
+    def _read_file(self, file_path: Path) -> dict[str, Any]:
+        file_contents = file_path.read_text()
+        return toml.loads(file_contents)
+
+    def __call__(self):
+        """Make the file reading at the source instantiation."""
+        self.init_kwargs = self._read_files(self.toml_file_path)
+        return super().__call__()
+
 
-    This can parse JSON, TOML, and YAML files based on the extensions.
+class YamlConfigSettingsSource(_YamlConfigSettingsSource):
+    """Use the nskit.common.io.yaml loading to load settings from a yaml file."""
+    def _read_file(self, file_path: Path) -> dict[str, Any]:
+        encoding = self.yaml_file_encoding or 'utf-8'
+        file_contents = file_path.read_text(encoding)
+        return yaml.loads(file_contents)
+
+    def __call__(self):
+        """Make the file reading at the source instantiation."""
+        self.init_kwargs = self._read_files(self.yaml_file_path)
+        return super().__call__()
+
+
+class DotEnvSettingsSource(_DotEnvSettingsSource):
+    """Fixes change of behaviour in pydantic-settings 2.2.0 with extra allowed handling.
+
+    Adds dotenv_extra variable that is set to replicate previous behaviour (ignore).
     """
 
-    def __init__(self, *args, **kwargs):
-        """Initialise the Settings Source."""
-        super().__init__(*args, **kwargs)
-        self.__parsed_contents = None
-
-    def get_field_value(
-        self, field: FieldInfo, field_name: str  # noqa: U100
-    ) -> Tuple[Any, str, bool]:
-        """Get a field value."""
-        if self.__parsed_contents is None:
-            try:
-                encoding = self.config.get('env_file_encoding', 'utf-8')
-                file_path = Path(self.config.get('config_file_path'))
-                file_type = self.config.get('config_file_type', None)
-                file_contents = file_path.read_text(encoding)
-                if file_path.suffix.lower() in ['.jsn', '.json'] or (file_type is not None and file_type.lower() == 'json'):
-                    self.__parsed_contents = json.loads(file_contents)
-                elif file_path.suffix.lower() in ['.tml', '.toml'] or (file_type is not None and file_type.lower() == 'toml'):
-                    self.__parsed_contents = toml.loads(file_contents)
-                elif file_path.suffix.lower() in ['.yml', '.yaml'] or (file_type is not None and file_type.lower() == 'yaml'):
-                    self.__parsed_contents = yaml.loads(file_contents)
-            except Exception:
-                pass  # nosec B110
-        if self.__parsed_contents is not None:
-            field_value = self.__parsed_contents.get(field_name)
-        else:
-            field_value = None
-        return field_value, field_name, False
-
-    def prepare_field_value(
-        self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool  # noqa: U100
-    ) -> Any:
-        """Prepare the field value."""
-        return value
-
-    def __call__(self) -> Dict[str, Any]:
-        """Call the source."""
-        d: Dict[str, Any] = {}
-
-        for field_name, field in self.settings_cls.model_fields.items():
-            field_value, field_key, value_is_complex = self.get_field_value(
-                field, field_name
-            )
-            field_value = self.prepare_field_value(
-                field_name, field, field_value, value_is_complex
-            )
-            if field_value is not None:
-                d[field_key] = field_value
-
-        return d
-
-    def _load_file(self, file_path: Path, encoding: str) -> Dict[str, Any]:  # noqa: U100
-        file_path = Path(file_path)
+    def __init__(
+        self,
+        settings_cls: type[BaseSettings],
+        env_file: DotenvType | None = ENV_FILE_SENTINEL,
+        env_file_encoding: str | None = None,
+        case_sensitive: bool | None = None,
+        env_prefix: str | None = None,
+        env_nested_delimiter: str | None = None,
+        env_ignore_empty: bool | None = None,
+        env_parse_none_str: str | None = None,
+        dotenv_extra: ExtraValues | None = 'ignore'
+    ) -> None:
+        """Wrapper for init function to add dotenv_extra handling."""
+        self.dotenv_extra = dotenv_extra
+        super().__init__(
+            settings_cls,
+            env_file,
+            env_file_encoding,
+            case_sensitive,
+            env_prefix,
+            env_nested_delimiter,
+            env_ignore_empty,
+            env_parse_none_str
+        )
+
+    def __call__(self) -> dict[str, Any]:
+        """Wraps call logic introduced in 2.2.0, but is backwards compatible to 2.1.0 and earlier versions."""
+        data: dict[str, Any] = super().__call__()
+        to_pop = []
+        for key in data.keys():
+            matched = False
+            for field_name, field in self.settings_cls.model_fields.items():
+                for field_alias, field_env_name, _ in self._extract_field_info(field, field_name):
+                    if key == field_env_name or key == field_alias:
+                        matched = True
+                        break
+                if matched:
+                    break
+            if not matched and self.dotenv_extra == 'ignore':
+                to_pop.append(key)
+        for key in to_pop:
+            data.pop(key)
+        return data

src/nskit/vcs/providers/azure_devops.py

@@ -9,22 +9,17 @@
     raise ImportError('Azure Devops Provider requires installing extra dependencies, use pip install nskit[azure_devops]')
 
 from pydantic import HttpUrl
-from pydantic_settings import SettingsConfigDict
 
+from nskit.common.configuration import SettingsConfigDict
 from nskit.vcs.providers.abstract import RepoClient, VCSProviderSettings
 
 # Want to use MS interactive Auth as default, but can't get it working, instead, using cli invoke
 
 
 class AzureDevOpsSettings(VCSProviderSettings):
     """Azure DevOps settings."""
-    # This is not ideal behaviour, but due to the issue highlighted in
-    # https://github.com/pydantic/pydantic-settings/issues/245 and the
-    # non-semver compliant versioning in pydantic-settings, we need to add this behaviour
-    # this now changes the API behaviour for these objects as they will
-    # also ignore additional inputs in the python initialisation
-    #  We will pin to version < 2.1.0 instead of allowing 2.2.0+ as it requires the code below:
-    model_config = SettingsConfigDict(env_prefix='AZURE_DEVOPS_', env_file='.env')  # , extra='ignore')  noqa: E800
+    model_config = SettingsConfigDict(env_prefix='AZURE_DEVOPS_', env_file='.env', dotenv_extra='ignore')
+
     url: HttpUrl = "https://dev.azure.com"
     organisation: str
     project: str

src/nskit/vcs/providers/github.py

@@ -9,15 +9,14 @@
 except ImportError:
     raise ImportError('Github Provider requires installing extra dependencies (ghapi), use pip install nskit[github]')
 from pydantic import Field, field_validator, HttpUrl, SecretStr, ValidationInfo
-from pydantic_settings import SettingsConfigDict
 
-from nskit.common.configuration import BaseConfiguration
+from nskit.common.configuration import BaseConfiguration, SettingsConfigDict
 from nskit.vcs.providers.abstract import RepoClient, VCSProviderSettings
 
 
 class GithubRepoSettings(BaseConfiguration):
     """Github Repo settings."""
-    model_config = SettingsConfigDict(env_prefix='GITHUB_REPO', env_file='.env')
+    model_config = SettingsConfigDict(env_prefix='GITHUB_REPO_', env_file='.env', dotenv_extra='ignore')
 
     private: bool = True
     has_issues: Optional[bool] = None
@@ -36,13 +35,8 @@ class GithubSettings(VCSProviderSettings):
 
     Uses PAT token for auth (set in environment variables as GITHUB_TOKEN)
     """
-    # This is not ideal behaviour, but due to the issue highlighted in
-    # https://github.com/pydantic/pydantic-settings/issues/245 and the
-    # non-semver compliant versioning in pydantic-settings, we need to add this behaviour
-    # this now changes the API behaviour for these objects as they will
-    # also ignore additional inputs in the python initialisation
-    # We will pin to version < 2.1.0 instead of allowing 2.2.0+ as it requires the code below:
-    model_config = SettingsConfigDict(env_prefix='GITHUB_', env_file='.env')  # extra='ignore')  noqa: E800
+    model_config = SettingsConfigDict(env_prefix='GITHUB_', env_file='.env', dotenv_extra='ignore')
+
     interactive: bool = Field(False, description='Use Interactive Validation for token')
     url: HttpUrl = "https://api.github.com"
     organisation: Optional[str] = Field(None, description='Organisation to work in, otherwise uses the user for the token')