BREAKING CHANGE: update pydantic version and use settings class inste…

…ad of config

README.md

@@ -1,24 +1,22 @@
 # pydantic-ssm-settings
 
-Replace Pydantic's builtin [Secret Support](https://pydantic-docs.helpmanual.io/usage/settings/#secret-support) with a configuration provider that loads parameters from [AWS Systems Manager Parameter Store](https://docs.aws.amazon.com/systems-manager/latest/userguide/systems-manager-parameter-store.html). Parameters are loaded _lazily_, meaning that they are only requested from AWS if they are not provided via [standard field value priority](https://pydantic-docs.helpmanual.io/usage/settings/#field-value-priority) (i.e. initialiser, environment variable, or via `.env` file).
+Replace Pydantic's builtin [Secret Support](https://pydantic-docs.helpmanual.io/usage/settings/#secret-support) with a configuration provider that loads parameters from [AWS Systems Manager Parameter Store](https://docs.aws.amazon.com/systems-manager/latest/userguide/systems-manager-parameter-store.html). Parameters are loaded _lazily_, meaning that they are only requested from AWS if they are not provided via [standard field value priority](https://pydantic-docs.helpmanual.io/usage/settings/#field-value-priority) (i.e. initializer, environment variable, or via `.env` file).
 
 ## Usage
 
-The simplest way to use this module is to inhert your settings `Config` class from `AwsSsmSourceConfig`. This will overwrite the [`file_secret_settings` settings source](https://pydantic-docs.helpmanual.io/usage/settings/#customise-settings-sources) with the `AwsSsmSettingsSource`. Provide a prefix to SSM parameters via the `_secrets_dir` initialiser value or the `secrets_dir` Config value.
+The simplest way to use this module is to inhert your settings class from `AwsSsmSettings`. This will overwrite the [`pydantic` settings sources order](https://docs.pydantic.dev/latest/concepts/pydantic_settings/#customise-settings-sources) with the `AwsSsmSettingsSource`. Provide a prefix to SSM parameters via the `_secrets_dir` initializer value or the `secrets_dir` Config value.
 
 ```py
 from pydantic import BaseSettings
-from pydantic_ssm_settings import AwsSsmSourceConfig
+from pydantic_ssm_settings import AwsSsmSettings
 
 
-class WebserviceSettings(BaseSettings):
+class WebserviceSettings(AwsSsmSettings):
     some_val: str
     another_val: int
 
-    class Config(AwsSsmSourceConfig):
-        ...
 
 SimpleSettings(_secrets_dir='/prod/webservice')
 ```
 
-The above example will attempt to retreive values from `/prod/webservice/some_val` and `/prod/webservice/another_val` if not provided otherwise.
+The above example will attempt to retrieve values from `/prod/webservice/some_val` and `/prod/webservice/another_val` if not provided otherwise.

pydantic_ssm_settings/__init__.py

@@ -1,4 +1,4 @@
-from .settings import AwsSsmSourceConfig
+from .settings import AwsSsmSettings
 
-__all__ = ("AwsSsmSourceConfig",)
+__all__ = ("AwsSsmSettings",)
 __version__ = "0.2.4"

pydantic_ssm_settings/settings.py

@@ -1,30 +1,28 @@
 import logging
-from typing import Tuple
+from typing import Tuple, Type
 
-from pydantic.env_settings import (
-    EnvSettingsSource,
-    InitSettingsSource,
-    SecretsSettingsSource,
-    SettingsSourceCallable,
-)
-
-from .source import AwsSsmSettingsSource
+from pydantic_settings import BaseSettings, PydanticBaseSettingsSource
+from pydantic_ssm_settings.source import AwsSsmSettingsSource
 
 logger = logging.getLogger(__name__)
 
 
-class AwsSsmSourceConfig:
+class AwsSsmSettings(BaseSettings):
+
     @classmethod
-    def customise_sources(
+    def settings_customise_sources(
         cls,
-        init_settings: InitSettingsSource,
-        env_settings: EnvSettingsSource,
-        file_secret_settings: SecretsSettingsSource,
-    ) -> Tuple[SettingsSourceCallable, ...]:
+        settings_cls: Type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> Tuple[PydanticBaseSettingsSource, ...]:
 
         ssm_settings = AwsSsmSettingsSource(
-            ssm_prefix=file_secret_settings.secrets_dir,
-            env_nested_delimiter=env_settings.env_nested_delimiter,
+            settings_cls=settings_cls,
+            ssm_prefix=settings_cls.model_config.get("secrets_dir"),
+            env_nested_delimiter=settings_cls.model_config.get("env_nested_delimiter"),
         )
 
         return (

pydantic_ssm_settings/source.py

@@ -1,16 +1,13 @@
 import os
 import logging
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Tuple
+from typing import TYPE_CHECKING, Mapping, Optional, Union
 
 from botocore.exceptions import ClientError
 from botocore.client import Config
 import boto3
 
-from pydantic import BaseSettings
-from pydantic.typing import StrPath, get_origin, is_union
-from pydantic.utils import deep_update
-from pydantic.fields import ModelField
+from pydantic_settings import BaseSettings, EnvSettingsSource
 
 if TYPE_CHECKING:
     from mypy_boto3_ssm.client import SSMClient
@@ -19,20 +16,28 @@
 logger = logging.getLogger(__name__)
 
 
-class SettingsError(ValueError):
-    pass
+class AwsSsmSettingsSource(EnvSettingsSource):
+    """
+    A simple settings source class that loads variables from a JSON file
+    at the project's root.
 
-
-class AwsSsmSettingsSource:
-    __slots__ = ("ssm_prefix", "env_nested_delimiter")
+    Here we happen to choose to use the `env_file_encoding` from Config
+    when reading `config.json`
+    """
 
     def __init__(
         self,
-        ssm_prefix: Optional[StrPath],
+        settings_cls: type[BaseSettings],
+        ssm_prefix: Optional[Union[str, Path]] = None,
+        case_sensitive: Optional[bool] = None,
+        env_prefix: Optional[str] = None,
         env_nested_delimiter: Optional[str] = None,
-    ):
-        self.ssm_prefix: Optional[StrPath] = ssm_prefix
-        self.env_nested_delimiter: Optional[str] = env_nested_delimiter
+    ) -> None:
+        self.ssm_prefix = ssm_prefix if ssm_prefix is not None else settings_cls.model_config.get('secrets_dir')
+        super().__init__(settings_cls, case_sensitive, env_prefix, env_nested_delimiter)
+
+    def __repr__(self) -> str:
+        return f"AwsSsmSettingsSource(ssm_prefix={self.ssm_prefix!r})"
 
     @property
     def client(self) -> "SSMClient":
@@ -43,124 +48,29 @@ def client_config(self) -> Config:
         timeout = float(os.environ.get("SSM_TIMEOUT", 0.5))
         return Config(connect_timeout=timeout, read_timeout=timeout)
 
-    def load_from_ssm(self, secrets_path: Path, case_sensitive: bool):
+    def _load_env_vars(self) -> Mapping[str, Optional[str]]:
+        logger.debug(f"Building SSM settings with prefix of {str(self.ssm_prefix)}")
 
-        if not secrets_path.is_absolute():
-            raise ValueError("SSM prefix must be absolute path")
+        output: Mapping[str, Optional[str]] = {}
+        if self.ssm_prefix is None:
+            return output
 
-        logger.debug(f"Building SSM settings with prefix of {secrets_path=}")
+        self.ssm_path = Path(self.ssm_prefix).expanduser()
+        if not self.ssm_path.is_absolute():
+            raise ValueError("SSM prefix must be absolute path")
 
-        output = {}
         try:
             paginator = self.client.get_paginator("get_parameters_by_path")
             response_iterator = paginator.paginate(
-                Path=str(secrets_path), WithDecryption=True
+                Path=str(self.ssm_path), WithDecryption=True
             )
 
             for page in response_iterator:
                 for parameter in page["Parameters"]:
-                    key = Path(parameter["Name"]).relative_to(secrets_path).as_posix()
-                    output[key if case_sensitive else key.lower()] = parameter["Value"]
+                    key = Path(parameter["Name"]).relative_to(self.ssm_path).as_posix()
+                    output[key if self.case_sensitive else key.lower()] = parameter["Value"]
 
         except ClientError:
-            logger.exception("Failed to get parameters from %s", secrets_path)
+            logger.exception("Failed to get parameters from %s", str(self.ssm_path))
 
         return output
-
-    def __call__(self, settings: BaseSettings) -> Dict[str, Any]:
-        """
-        Returns SSM values for all settings.
-        """
-        d: Dict[str, Optional[Any]] = {}
-
-        if self.ssm_prefix is None:
-            return d
-
-        ssm_values = self.load_from_ssm(
-            secrets_path=Path(self.ssm_prefix),
-            case_sensitive=settings.__config__.case_sensitive,
-        )
-
-        # The following was lifted from https://github.com/samuelcolvin/pydantic/blob/a21f0763ee877f0c86f254a5d60f70b1002faa68/pydantic/env_settings.py#L165-L237  # noqa
-        for field in settings.__fields__.values():
-            env_val: Optional[str] = None
-            for env_name in field.field_info.extra["env_names"]:
-                env_val = ssm_values.get(env_name)
-                if env_val is not None:
-                    break
-
-            is_complex, allow_json_failure = self._field_is_complex(field)
-            if is_complex:
-                if env_val is None:
-                    # field is complex but no value found so far, try explode_env_vars
-                    env_val_built = self._explode_ssm_values(field, ssm_values)
-                    if env_val_built:
-                        d[field.alias] = env_val_built
-                else:
-                    # field is complex and there's a value, decode that as JSON, then
-                    # add explode_env_vars
-                    try:
-                        env_val = settings.__config__.json_loads(env_val)
-                    except ValueError as e:
-                        if not allow_json_failure:
-                            raise SettingsError(
-                                f'error parsing JSON for "{env_name}"'
-                            ) from e
-
-                    if isinstance(env_val, dict):
-                        d[field.alias] = deep_update(
-                            env_val, self._explode_ssm_values(field, ssm_values)
-                        )
-                    else:
-                        d[field.alias] = env_val
-            elif env_val is not None:
-                # simplest case, field is not complex, we only need to add the
-                # value if it was found
-                d[field.alias] = env_val
-
-        return d
-
-    def _field_is_complex(self, field: ModelField) -> Tuple[bool, bool]:
-        """
-        Find out if a field is complex, and if so whether JSON errors should be ignored
-        """
-        if field.is_complex():
-            allow_json_failure = False
-        elif (
-            is_union(get_origin(field.type_))
-            and field.sub_fields
-            and any(f.is_complex() for f in field.sub_fields)
-        ):
-            allow_json_failure = True
-        else:
-            return False, False
-
-        return True, allow_json_failure
-
-    def _explode_ssm_values(
-        self, field: ModelField, env_vars: Mapping[str, Optional[str]]
-    ) -> Dict[str, Any]:
-        """
-        Process env_vars and extract the values of keys containing
-        env_nested_delimiter into nested dictionaries.
-
-        This is applied to a single field, hence filtering by env_var prefix.
-        """
-        prefixes = [
-            f"{env_name}{self.env_nested_delimiter}"
-            for env_name in field.field_info.extra["env_names"]
-        ]
-        result: Dict[str, Any] = {}
-        for env_name, env_val in env_vars.items():
-            if not any(env_name.startswith(prefix) for prefix in prefixes):
-                continue
-            _, *keys, last_key = env_name.split(self.env_nested_delimiter)
-            env_var = result
-            for key in keys:
-                env_var = env_var.setdefault(key, {})
-            env_var[last_key] = env_val
-
-        return result
-
-    def __repr__(self) -> str:
-        return f"AwsSsmSettingsSource(ssm_prefix={self.ssm_prefix!r})"

pyproject.toml

@@ -14,9 +14,10 @@ repository = "https://github.com/developmentseed/pydantic-ssm-settings/"
 version = "0.2.4"
 
 [tool.poetry.dependencies]
-"boto3" = "^1.21.45"
-pydantic = "^1.6.2"
-python = "^3.7"
+boto3 = "^1.21.45"
+pydantic = "^2.4.2"
+pydantic-settings = "^2.0"
+python = "^3.8"
 
 [tool.poetry.dev-dependencies]
 black = "^22.3.0"
@@ -32,12 +33,12 @@ python-semantic-release = "^7.32.0"
 
 [tool.semantic_release]
 # https://python-semantic-release.readthedocs.io/en/latest/configuration.html
-branch = "main" 
+branch = "main"
 build_command = "pip install poetry && poetry build" # https://github.com/relekang/python-semantic-release/issues/222#issuecomment-709326972
-upload_to_pypi = true 
-upload_to_release = true 
-version_toml = "pyproject.toml:tool.poetry.version" 
-version_variable = "pydantic_ssm_settings/__init__.py:__version__" 
+upload_to_pypi = true
+upload_to_release = true
+version_toml = "pyproject.toml:tool.poetry.version"
+version_variable = "pydantic_ssm_settings/__init__.py:__version__"
 
 [tool.flake8]
 ignore = ["E203", "W503"]

tests/test_main.py

@@ -2,27 +2,20 @@
 
 import pytest
 
-from pydantic import BaseSettings
-from pydantic_ssm_settings import AwsSsmSourceConfig
+from pydantic_ssm_settings import AwsSsmSettings
 
 logger = logging.getLogger("pydantic_ssm_settings")
 logger.setLevel(logging.DEBUG)
 
 
-class SimpleSettings(BaseSettings):
+class SimpleSettings(AwsSsmSettings):
     foo: str
 
-    class Config(AwsSsmSourceConfig):
-        ...
 
-
-class IntSettings(BaseSettings):
+class IntSettings(AwsSsmSettings):
     foo: str
     bar: int
 
-    class Config(AwsSsmSourceConfig):
-        ...
-
 
 def test_secrets_dir_must_be_absolute():
     with pytest.raises(ValueError):