Feature/mx 1596 simplify settings handling (#220)

# PR Context
<!-- Additional info for the reviewer -->

# Added
<!-- New features and interfaces -->

# Changes
<!-- Changes in existing functionality -->

# Deprecated
<!-- Soon-to-be removed features -->

# Removed
<!-- Definitely removed features -->

- BREAKING: ability to store different settings instances at the same
time. Dependent
    repositories now must bundle all settings in a single class.

# Fixed
<!-- Fixed bugs -->

# Security
<!-- Fixed vulnerabilities -->

CHANGELOG.md

@@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Deprecated
 
 ### Removed
+- BREAKING: ability to store different settings instances at the same time. Dependent
+    repositories now must bundle all settings in a single class.
 
 ### Fixed
 

mex/common/context.py

@@ -32,3 +32,36 @@ def pop(self, cls: type[_SingletonT]) -> _SingletonT:
     def reset(self) -> None:
         """Remove all singleton instances from the store."""
         self._instances_by_class.clear()
+
+
+class SingleSingletonStore(Generic[_SingletonT]):
+    """Thin wrapper for storing a single thread-local singleton.
+
+    Stores only a single instance. Requested class must either be
+    the same or a parent class of the stored class.
+    """
+
+    def __init__(self) -> None:
+        """Create a new settings singleton store."""
+        self._singleton: _SingletonT | None = None
+
+    def load(self, cls: type[_SingletonT]) -> _SingletonT:
+        """Retrieve the settings for the given class or create a new one."""
+        if self._singleton is None:
+            self._singleton = cls()
+            return self._singleton
+        if not issubclass(type(self._singleton), cls):
+            raise RuntimeError(
+                f"requested class ({cls}) is not a parent class of loaded class "
+                f"({type(self._singleton)}). "
+                f"Did you initialize {cls} upon startup?"
+            )
+        return self._singleton
+
+    def push(self, instance: _SingletonT) -> None:
+        """Set or replace a singleton instance in the store."""
+        self._singleton = instance
+
+    def reset(self) -> None:
+        """Remove singleton instance from the store."""
+        self._singleton = None

mex/common/settings.py

@@ -2,15 +2,16 @@
 from typing import Any, Self, cast
 
 from pydantic import AnyUrl, Field, SecretStr, model_validator
+from pydantic import BaseModel as PydanticBaseModel
 from pydantic_core import Url
 from pydantic_settings import BaseSettings as PydanticBaseSettings
 from pydantic_settings import SettingsConfigDict
 from pydantic_settings.sources import ENV_FILE_SENTINEL, DotenvType, EnvSettingsSource
 
-from mex.common.context import SingletonStore
+from mex.common.context import SingleSingletonStore
 from mex.common.types import AssetsPath, IdentityProvider, Sink, WorkPath
 
-SETTINGS_STORE = SingletonStore["BaseSettings"]()
+SETTINGS_STORE = SingleSingletonStore["BaseSettings"]()
 
 
 class BaseSettings(PydanticBaseSettings):
@@ -34,6 +35,7 @@ class BaseSettings(PydanticBaseSettings):
         env_prefix="mex_",
         env_file=".env",
         env_file_encoding="utf-8",
+        env_nested_delimiter="__",
         extra="ignore",
         validate_default=True,
         validate_assignment=True,
@@ -206,33 +208,17 @@ def get_env_name(cls, name: str) -> str:
     @model_validator(mode="after")
     def resolve_paths(self) -> Self:
         """Resolve AssetPath and WorkPath."""
-        for name in self.model_fields:
-            value = getattr(self, name)
+
+        def _resolve(model: PydanticBaseModel, _name: str) -> None:
+            value = getattr(model, _name)
             if isinstance(value, AssetsPath) and value.is_relative():
-                setattr(self, name, self.assets_dir.resolve() / value)
+                setattr(model, _name, self.assets_dir.resolve() / value)
             elif isinstance(value, WorkPath) and value.is_relative():
-                setattr(self, name, self.work_dir.resolve() / value)
-        return self
-
-    @model_validator(mode="after")
-    def sync_settings(self) -> Self:
-        """Sync updates to settings in the `BaseSettings` scope to other settings.
+                setattr(model, _name, self.work_dir.resolve() / value)
+            elif isinstance(value, PydanticBaseModel):
+                for sub_model_field_name in value.model_fields:
+                    _resolve(value, sub_model_field_name)
 
-        Caveat: Updating fields of one class does not automatically update other
-        classes. To update another class, call any settings `.get()` method.
-        """
-        # ensure the settings singled instance is stored
-        SETTINGS_STORE.push(self)
-        # collect the changes to fields in the `BaseSettings` scope
-        base_scope = {
-            field: value
-            for field, value in self.model_dump(exclude_unset=True).items()
-            if field in BaseSettings.model_fields
-        }
-        # iterate over all active setting instances and swap them out
-        for settings in SETTINGS_STORE:
-            # create a new setting instance with `BaseSettings` fields are overwritten
-            SETTINGS_STORE.push(
-                settings.model_construct(**{**settings.model_dump(), **base_scope})
-            )
-        return cast(Self, SETTINGS_STORE.load(type(self)))
+        for name in self.model_fields:
+            _resolve(self, name)
+        return self

tests/identity/test_registry.py

@@ -11,7 +11,7 @@
     get_provider,
     register_provider,
 )
-from mex.common.settings import BaseSettings
+from mex.common.settings import SETTINGS_STORE, BaseSettings
 from mex.common.types import IdentityProvider
 
 
@@ -60,6 +60,7 @@ def test_get_provider_error(monkeypatch: MonkeyPatch) -> None:
 
 
 def test_get_provider() -> None:
+    SETTINGS_STORE.reset()
     settings = DummySettings.get()
     settings.identity_provider = DummyIdentityProvider.DUMMY
 

tests/test_context.py

@@ -0,0 +1,31 @@
+import pytest
+
+from mex.common.context import SingleSingletonStore
+
+
+class Parent:
+    pass
+
+
+class Child(Parent):
+    pass
+
+
+def test_single_singleton_store() -> None:
+    store = SingleSingletonStore["Parent"]()
+    parent = store.load(Parent)
+
+    with pytest.raises(RuntimeError, match="is not a parent class of loaded class"):
+        store.load(Child)
+
+    assert parent is store.load(Parent)
+
+    store.push(Child())
+
+    child = store.load(Child)
+
+    assert child is store.load(Parent)
+
+    store.reset()
+
+    assert store._singleton is None

tests/test_settings.py

@@ -4,6 +4,7 @@
 
 import pytest
 
+from mex.common.models import BaseModel
 from mex.common.settings import SETTINGS_STORE, BaseSettings
 from mex.common.types import AssetsPath, WorkPath
 
@@ -60,11 +61,16 @@ def test_parse_env_file() -> None:
 
 
 def test_resolve_paths() -> None:
+
+    class SubModel(BaseModel):
+        sub_model_path: WorkPath
+
     class DummySettings(BaseSettings):
         non_path: str
         abs_work_path: WorkPath
         rel_work_path: WorkPath
         assets_path: AssetsPath
+        sub_model: SubModel
 
     if platform.system() == "Windows":  # pragma: no cover
         absolute = WorkPath(r"C:\absolute\path")
@@ -80,58 +86,16 @@ class DummySettings(BaseSettings):
         assets_path=AssetsPath(relative),
         assets_dir=Path(absolute / "assets_dir"),
         work_dir=Path(absolute / "work_dir"),
+        sub_model=SubModel(sub_model_path=relative),
     )
 
-    settings_dict = DummySettings.get().model_dump(exclude_defaults=True)
+    settings_dict = settings.model_dump()
     assert settings_dict["non_path"] == "blablabla"
     assert settings_dict["abs_work_path"] == absolute
     assert settings_dict["rel_work_path"] == WorkPath(settings.work_dir / relative)
     assert settings_dict["assets_path"] == AssetsPath(
         absolute / "assets_dir" / relative
     )
-
-
-class BlueSettings(BaseSettings):
-    color: str = "blue"
-
-
-class RedSettings(BaseSettings):
-    color: str = "red"
-
-
-def test_sync_settings_from_base(tmp_path: Path) -> None:
-    # GIVEN an instance of the base settings and a subclass
-    base_settings = BaseSettings.get()
-    blue_settings = BlueSettings.get()
-
-    # GIVEN a field that belongs to the `BaseSettings` scope
-    assert "work_dir" in BaseSettings.model_fields
-
-    # GIVEN the two settings start out with the same `work_dir`
-    assert base_settings.work_dir == blue_settings.work_dir
-
-    # WHEN we change the `work_dir` on the `BaseSetting`
-    base_settings.work_dir = tmp_path / "base-update"
-
-    # THEN the changes should be synced to new `BlueSettings`
-    blue_settings = BlueSettings.get()
-    assert blue_settings.work_dir == tmp_path / "base-update"
-
-
-def test_sync_settings_from_subclasses(tmp_path: Path) -> None:
-    # GIVEN an instance of the base settings and two subclasses
-    base_settings = BaseSettings.get()
-    blue_settings = BlueSettings.get()
-    red_settings = RedSettings.get()
-
-    # GIVEN all settings start out with the same `work_dir`
-    assert base_settings.work_dir == blue_settings.work_dir == red_settings.work_dir
-
-    # WHEN we change the `work_dir` on the `BlueSetting`
-    blue_settings.work_dir = tmp_path / "blue-update"
-
-    # THEN the changes should be synced to new `BaseSettings` and `RedSettings`
-    base_settings = BaseSettings.get()
-    red_settings = RedSettings.get()
-    assert blue_settings.work_dir == tmp_path / "blue-update"
-    assert red_settings.work_dir == tmp_path / "blue-update"
+    assert settings_dict["sub_model"]["sub_model_path"] == WorkPath(
+        settings.work_dir / relative
+    )