Add support for instance-level sources with ellipsis hierarchy

resolution

Author: Евгений Блинов

README.md

@@ -330,15 +330,16 @@ Each field value is resolved in the following order:
 
 ```mermaid
 graph TD;
-  A[Default values] --> B(Data sources in the order listed) --> C(The values set at runtime)
+  A[Default values] --> B(Class sources) --> C(Field sources) --> D(Instance sources) --> E(The values set at runtime)
 ```
 
 That is, values obtained from sources have higher priority than default values, but can be overwritten (unless you [prohibit it](#read-only-fields)) by other values at runtime.
 
-There are two ways to specify a list of sources:
+There are three ways to specify a list of sources:
 
 - For the **whole class**.
 - For a **specific field**.
+- For a **specific instance**.
 
 To specify a list of sources for the entire class, pass it to the class constructor:
 
@@ -363,9 +364,37 @@ class MyClass(Storage, sources=[TOMLSource('pyproject.toml', table='tool.my_tool
     some_field = Field('some_value', sources=[TOMLSource('config_for_this_field.toml'), ...])
 ```
 
+Finally, you can specify a list of sources for a specific instance by passing it as the `_sources` argument when creating the object:
+
+```python
+instance = MyClass(_sources=[TOMLSource('instance_config.toml')])
+```
+
+Without an ellipsis, instance-level sources completely replace both class-level and field-level sources. If you want instance-level sources to have the highest priority while still falling back to other sources, use an ellipsis:
+
+```python
+instance = MyClass(_sources=[TOMLSource('instance_config.toml'), ...])
+```
+
+In this case, instance sources are checked first, and if a value is not found, the lookup falls back to the sources that the field would normally use without `_sources`. The fallback rules are:
+
+- If a field has no `sources` parameter → fallback to class-level sources directly.
+- If a field has `sources` without `...` → fallback to field-level sources only (class-level sources are **not** included).
+- If a field has `sources` with `...` → fallback to field-level sources, then class-level sources.
+
+> ⚠️ This means that `...` in `_sources` does **not** always reach class-level sources. If a field defines its own `sources` without `...`, class-level sources are excluded for that field even when instance-level `_sources` contains `...`:
+>
+> ```python
+> class MyClass(Storage, sources=[EnvSource()]):
+>     # This field's sources do not include ..., so EnvSource() is unreachable for it:
+>     some_field = Field('default', sources=[TOMLSource('field_config.toml')])
+> ```
+
+Only `list` and `tuple` are accepted as the `_sources` collection type.
+
 All values from sources are loaded when the config object is created. This means that (theoretically) during program execution, you can, for example, change a configuration file, then create a new storage object, and its contents will be different. The old object will not automatically know that the config file has been changed. Avoid this pattern, as it can lead to subtle bugs.
 
-Each data source behaves like a mapping, and field values are looked up by field name. If no value is found in any of the sources, only then will the default value be used. The order in which the contents of the sources are checked corresponds to the order in which the sources themselves are listed, with sources for a field having higher priority than sources for the class as a whole.
+Each data source behaves like a mapping, and field values are looked up by field name. If no value is found in any of the sources, only then will the default value be used. The order in which the contents of the sources are checked corresponds to the order in which the sources themselves are listed. When multiple levels of sources are combined via ellipsis, instance-level sources have the highest priority, followed by field-level sources, and then class-level sources.
 
 For any field, you can change the key used to search for its value in the sources using the `alias` parameter:
 

skelet/fields/base.py

@@ -1,3 +1,6 @@
+from collections.abc import Sequence
+from sys import version_info
+from threading import Lock
 from typing import (
     Any,
     Callable,
@@ -13,16 +16,6 @@
     get_type_hints,
 )
 
-# TODO: check, EllipsisType was added to types module in Python 3.10.
-try:
-    from types import EllipsisType  # type: ignore[attr-defined, unused-ignore]
-except ImportError:  # pragma: no cover
-    EllipsisType = type(...)  # type: ignore[misc, unused-ignore]
-
-from collections.abc import Sequence
-from sys import version_info
-from threading import Lock
-
 from denial import InnerNoneType
 from locklib import ContextLockProtocol
 from sigmatch import PossibleCallMatcher, SignatureMismatchError
@@ -31,6 +24,7 @@
 from skelet.sources.abstract import AbstractSource, ExpectedType
 from skelet.sources.collection import SourcesCollection
 from skelet.storage import Storage
+from skelet.types import EllipsisType
 
 ValueType = TypeVar('ValueType')
 
@@ -255,19 +249,40 @@ def raise_exception_in_storage(self, exception: BaseException, raising_on: bool)
         self.exception = exception
 
     def get_sources(self, instance: Storage) -> SourcesCollection[ExpectedType]:
+        instance_sources_raw = instance.__instance_sources__
+
+        if instance_sources_raw is None:
+            return self._get_normal_sources(instance)
+
+        has_ellipsis = False
+        instance_only: List[AbstractSource[ExpectedType]] = []
+        for source in instance_sources_raw:
+            if source is Ellipsis:
+                has_ellipsis = True
+            else:
+                instance_only.append(cast(AbstractSource[ExpectedType], source))
+
+        if not has_ellipsis:
+            return cast(SourcesCollection[ExpectedType], SourcesCollection(instance_only))
+
+        normal_sources: SourcesCollection[ExpectedType] = self._get_normal_sources(instance)
+        combined: List[AbstractSource[ExpectedType]] = list(instance_only) + list(normal_sources.sources)
+        return cast(SourcesCollection[ExpectedType], SourcesCollection(combined))
+
+    def _get_normal_sources(self, instance: Storage) -> SourcesCollection[ExpectedType]:
         if self.sources is None:
             return instance.__sources__
 
-        result = []
+        result: List[AbstractSource[ExpectedType]] = []
         there_is_ellipsis = False
 
         for source in self.sources:
             if source is Ellipsis:
                 there_is_ellipsis = True
             else:
-                result.append(source)
+                result.append(cast(AbstractSource[ExpectedType], source))
 
         if there_is_ellipsis:
-           result.extend(instance.__sources__.sources)
+            result.extend(instance.__sources__.sources)
 
         return cast(SourcesCollection[ExpectedType], SourcesCollection(result))

skelet/storage.py

@@ -1,4 +1,5 @@
 from collections import defaultdict
+from collections.abc import Sequence
 from threading import Lock
 from typing import Any, Dict, List, Optional, Tuple, Union
 
@@ -8,6 +9,7 @@
 
 from skelet.sources.abstract import AbstractSource, ExpectedType
 from skelet.sources.collection import SourcesCollection
+from skelet.types import InstanceSourceItem
 
 sentinel = InnerNoneType()
 
@@ -17,8 +19,23 @@ class Storage:
     __field_names__: Union[List[str], Tuple[str, ...]] = ()
     __reverse_conflicts__: Dict[str, List[str]]
     __sources__: SourcesCollection  # type: ignore[type-arg]
+    __instance_sources__: Optional[Sequence[InstanceSourceItem]]
+
+    @staticmethod
+    def _pop_and_validate_instance_sources(kwargs: Dict[str, Any]) -> Optional[Sequence['InstanceSourceItem']]:
+        raw = kwargs.pop('_sources', sentinel)
+        if raw is sentinel:
+            return None
+        if not isinstance(raw, (list, tuple)):
+            raise TypeError('_sources must be a list or a tuple.')
+        for item in raw:
+            if item is not Ellipsis and not isinstance(item, AbstractSource):
+                raise TypeError(f'Each element of _sources must be a source or Ellipsis, got {type(item).__name__}.')
+        return raw
 
     def __init__(self, **kwargs: Any) -> None:
+        self.__instance_sources__ = self._pop_and_validate_instance_sources(kwargs)
+
         self.__values__: Dict[str, Any] = {}
         self.__locks__ = {field_name: Lock() for field_name in self.__field_names__}
         deduplicated_fields = set(self.__field_names__)

skelet/types.py

@@ -0,0 +1,13 @@
+from typing import Any, Union
+
+from skelet.sources.abstract import AbstractSource
+
+__all__ = ['EllipsisType', 'InstanceSourceItem']
+
+# EllipsisType was added to the types module in Python 3.10.
+try:
+    from types import EllipsisType  # type: ignore[attr-defined, unused-ignore]
+except ImportError:  # pragma: no cover
+    EllipsisType = type(...)  # type: ignore[misc, unused-ignore]
+
+InstanceSourceItem = Union[AbstractSource[Any], EllipsisType]

tests/functions/test_asdict.py

@@ -41,3 +41,36 @@ class SomeClass(Storage, sources=[MemorySource({'field': 42, 'second_field': 43}
         second_field = Field()
 
     assert asdict(SomeClass()) == {'field': 42, 'second_field': 43}
+
+
+@pytest.mark.parametrize('collection_type', [list, tuple])
+def test_instance_source_values_as_dict(collection_type):
+    class SomeClass(Storage):
+        field = Field(0)
+        second_field = Field(0)
+
+    instance = SomeClass(_sources=collection_type([MemorySource({'field': 42, 'second_field': 43})]))
+
+    assert asdict(instance) == {'field': 42, 'second_field': 43}
+
+
+@pytest.mark.parametrize('collection_type', [list, tuple])
+def test_instance_source_with_ellipsis_hierarchy_as_dict(collection_type):
+    class SomeClass(Storage, sources=[MemorySource({'class_field': 30})]):
+        instance_field = Field(0)
+        field_field = Field(0, sources=[MemorySource({'field_field': 20}), ...])
+        class_field = Field(0)
+
+    instance = SomeClass(_sources=collection_type([MemorySource({'instance_field': 10}), ...]))
+
+    assert asdict(instance) == {'instance_field': 10, 'field_field': 20, 'class_field': 30}
+
+
+@pytest.mark.parametrize('collection_type', [list, tuple])
+def test_instance_source_ellipsis_does_not_reach_class_when_field_has_no_ellipsis_as_dict(collection_type):
+    class SomeClass(Storage, sources=[MemorySource({'field': 99})]):
+        field = Field(0, sources=[MemorySource({'other': 1})])
+
+    instance = SomeClass(_sources=collection_type([MemorySource({'other': 2}), ...]))
+
+    assert asdict(instance) == {'field': 0}