0.0.16

.github/workflows/lint.yml

@@ -47,4 +47,4 @@ jobs:
 
       - name: Run mypy for tests
         shell: bash
-        run: mypy tests
+        run: mypy tests --exclude tests/typing

.gitignore

@@ -17,3 +17,6 @@ uv.lock
 .ropeproject
 node_modules
 mutants
+CLAUDE.md
+AGENTS.md
+.qwen

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "skelet"
-version = "0.0.15"
+version = "0.0.16"
 authors = [{ name = "Evgeniy Blinov", email = "zheni-b@yandex.ru" }]
 description = 'Collect all the settings in one place'
 readme = "README.md"
@@ -50,12 +50,13 @@ paths_to_mutate = "skelet"
 runner = "pytest"
 
 [tool.pytest.ini_options]
-markers = ["mypy_testing"]
+markers = ["mypy_testing", "thread_safety"]
 
 [tool.ruff]
 lint.ignore = ['E501', 'E712', 'PTH123', 'PTH118', 'PLR2004', 'PTH107', 'SIM105', 'SIM102', 'RET503', 'PLR0912', 'C901']
 lint.select = ["ERA001", "YTT", "ASYNC", "BLE", "B", "A", "COM", "INP", "PIE", "T20", "PT", "RSE", "RET", "SIM", "SLOT", "TID252", "ARG", "PTH", "I", "C90", "N", "E", "W", "D201", "D202", "D419", "F", "PL", "PLE", "PLR", "PLW", "RUF", "TRY201", "TRY400", "TRY401"]
 format.quote-style = "single"
+lint.per-file-ignores."tests/typing/**" = ["F821", "ARG001", "ARG005"]
 
 [project.urls]
 'Source' = 'https://github.com/mutating/skelet'

requirements_dev.txt

@@ -2,7 +2,8 @@ pytest==8.0.2
 coverage==7.6.1
 build==1.2.2.post1
 twine==6.1.0
-mypy==1.14.1
+mypy==1.14.1 ; python_version < "3.14"
+mypy==1.20.1 ; python_version >= "3.14"
 pytest-mypy-testing==0.1.3
 ruff==0.14.6
 mutmut==3.2.3

skelet/__init__.py

@@ -2,6 +2,7 @@
 from simtypes import NonNegativeInt as NonNegativeInt
 
 from skelet.fields.base import Field as Field
+from skelet.fields.base import FieldDescriptor as FieldDescriptor
 from skelet.functions.asdict import asdict as asdict
 from skelet.sources.cli import FixedCLISource as FixedCLISource
 from skelet.sources.env import EnvSource as EnvSource

skelet/fields/base.py

@@ -1,28 +1,20 @@
+from sys import version_info
+from threading import Lock
 from typing import (
     Any,
     Callable,
     Dict,
     Generic,
     List,
     Optional,
+    Sequence,
     Type,
-    TypeVar,
     Union,
     cast,
     get_origin,
     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,8 +23,7 @@
 from skelet.sources.abstract import AbstractSource, ExpectedType
 from skelet.sources.collection import SourcesCollection
 from skelet.storage import Storage
-
-ValueType = TypeVar('ValueType')
+from skelet.types import ChangeAction, EllipsisType, StorageType, ValueType
 
 if version_info < (3, 9):  # pragma: no cover
     SequenceWithStrings = Sequence
@@ -41,7 +32,7 @@
 
 sentinel = InnerNoneType()
 
-class Field(Generic[ValueType]):
+class FieldDescriptor(Generic[ValueType, StorageType]):
     def __init__(  # noqa: PLR0913, PLR0915
         self,
         default: Union[ValueType, InnerNoneType] = sentinel,
@@ -54,7 +45,7 @@ def __init__(  # noqa: PLR0913, PLR0915
         validation: Optional[Union[Dict[str, Callable[[ValueType], bool]], Callable[[ValueType], bool]]] = None,
         validate_default: bool = True,
         secret: bool = False,
-        action: Optional[Callable[[ValueType, ValueType, Storage], Any]] = None,
+        action: Optional[ChangeAction[ValueType, StorageType]] = None,
         read_lock: bool = False,
         conflicts: Optional[Dict[str, Callable[[ValueType, ValueType, Any, Any], bool]]] = None,
         reverse_conflicts: bool = True,
@@ -106,7 +97,7 @@ def __init__(  # noqa: PLR0913, PLR0915
         self.validation = validation
         self.validate_default = validate_default
         self.secret = secret
-        self.change_action = action
+        self.change_action: Optional[ChangeAction[ValueType, StorageType]] = action
         self.conflicts = conflicts
         self.reverse_conflicts_on = reverse_conflicts
         self.conversion = conversion
@@ -167,7 +158,7 @@ def locked_get(self, instance: Storage, instance_class: Type[Storage]) -> ValueT
     def unlocked_get(self, instance: Storage, instance_class: Type[Storage]) -> ValueType:  #noqa: ARG002
         return cast(ValueType, instance.__values__.get(cast(str, self.name)))
 
-    def __set__(self, instance: Storage, value: ValueType) -> None:
+    def __set__(self, instance: StorageType, value: ValueType) -> None:
         if self.read_only:
             raise AttributeError(f'{self.get_field_name_representation()} is read-only.')
 
@@ -223,9 +214,9 @@ def set_field_names(self, owner: Type[Storage], name: str) -> None:
             cast(List[str], owner.__field_names__).append(name)
 
     def check_type_hints(self, value: ValueType, strict: bool = False, raise_all: bool = False) -> None:
-        if not check(value, self.type_hint, strict=strict):  # type: ignore[arg-type]
+        if not check(value, self.type_hint, strict=strict):  # type: ignore[arg-type, unused-ignore]
             origin = get_origin(self.type_hint)
-            type_hint_name = self.type_hint.__name__ if origin is None else origin.__name__ if hasattr(origin, '__name__') else repr(origin)  # type: ignore[attr-defined]
+            type_hint_name = self.type_hint.__name__ if origin is None else origin.__name__ if hasattr(origin, '__name__') else repr(origin)  # type: ignore[attr-defined, unused-ignore]
             self.raise_exception_in_storage(TypeError(f'The value {self.get_value_representation(value)} of the {self.get_field_name_representation()} does not match the type {type_hint_name}.'), raise_all)
 
     def get_field_name_representation(self) -> str:
@@ -255,19 +246,77 @@ 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))
+
+
+def Field(  # noqa: PLR0913, N802
+    default: Any = sentinel,
+    /,
+    default_factory: Optional[Callable[[], Any]] = None,
+    doc: Optional[str] = None,
+    alias: Optional[str] = None,
+    sources: Optional[List[Union[AbstractSource[ExpectedType], EllipsisType]]] = None,
+    read_only: bool = False,
+    validation: Optional[Union[Dict[str, Callable[[Any], bool]], Callable[[Any], bool]]] = None,
+    validate_default: bool = True,
+    secret: bool = False,
+    action: Optional[ChangeAction[Any, StorageType]] = None,
+    read_lock: bool = False,
+    conflicts: Optional[Dict[str, Callable[[Any, Any, Any, Any], bool]]] = None,
+    reverse_conflicts: bool = True,
+    conversion: Optional[Callable[[Any], Any]] = None,
+    share_mutex_with: Optional[Sequence[str]] = None,
+) -> Any:
+    return FieldDescriptor(
+        default,
+        default_factory=default_factory,
+        doc=doc,
+        alias=alias,
+        sources=sources,
+        read_only=read_only,
+        validation=validation,
+        validate_default=validate_default,
+        secret=secret,
+        action=action,
+        read_lock=read_lock,
+        conflicts=conflicts,
+        reverse_conflicts=reverse_conflicts,
+        conversion=conversion,
+        share_mutex_with=share_mutex_with,
+    )

skelet/sources/abstract.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import Generic, Optional, Type, TypeVar, cast
+from typing import Generic, Optional, Type, TypeVar, Union, cast
 
 from denial import InnerNoneType
 from simtypes import check
@@ -12,22 +12,23 @@ class AbstractSource(Generic[ExpectedType], ABC):
     def __getitem__(self, key: str) -> ExpectedType:
         ...  # pragma: no cover
 
-    def get(self, key: str, default: Optional[ExpectedType] = None) -> Optional[ExpectedType]:
+    def get(self, key: str, default: Union[ExpectedType, InnerNoneType, None] = None) -> Union[ExpectedType, InnerNoneType, None]:
         try:
             result: ExpectedType = self[key]
         except KeyError:
             return default
 
         return result
 
-    def type_awared_get(self, key: str, hint: Type[ExpectedType], default: ExpectedType = cast(ExpectedType, sentinel)) -> Optional[ExpectedType]:  # noqa: B008
+    def type_awared_get(self, key: str, hint: Type[ExpectedType], default: Union[ExpectedType, InnerNoneType] = sentinel) -> Optional[ExpectedType]:
         result = self.get(key, default)
 
         if result is default:
             if default is sentinel:
                 return None
-            return default
+            return cast(ExpectedType, default)
 
+        result = cast(ExpectedType, result)
         if not check(result, hint, strict=True):
             raise TypeError(f'The value of the "{key}" field did not pass the type check.')
 

skelet/sources/cli.py

@@ -66,7 +66,7 @@ def __getitem__(self, key: str) -> str:  # type: ignore[override]
             except SystemExit as e:
                 raise KeyError(key) from e  # pragma: no cover
 
-    def type_awared_get(self, key: str, hint: Type[ExpectedType], default: ExpectedType = cast(ExpectedType, sentinel)) -> Optional[ExpectedType]:  # noqa: B008
+    def type_awared_get(self, key: str, hint: Type[ExpectedType], default: Union[ExpectedType, InnerNoneType] = sentinel) -> Optional[ExpectedType]:
         subresult = cast(Union[str, SentinelType], self.get(key, default))
 
         if hint is bool and key in self.named_arguments:
@@ -77,9 +77,9 @@ def type_awared_get(self, key: str, hint: Type[ExpectedType], default: ExpectedT
             raise CLIFormatError("You can't pass values for boolean named fields to the CLI.")
 
         if subresult is default:
-            if default is not sentinel:
-                return default
-            return None
+            if default is sentinel:
+                return None
+            return cast(ExpectedType, default)
 
         return from_string(cast(str, subresult), hint)
 

skelet/sources/collection.py

@@ -1,4 +1,4 @@
-from typing import Any, List, Optional, Type, cast
+from typing import Any, List, Optional, Type, Union, cast
 
 from denial import InnerNoneType
 from printo import repred
@@ -27,13 +27,13 @@ def get(self, key: str, default: Any = None) -> Any:
         except KeyError:
             return default
 
-    def type_awared_get(self, key: str, hint: Type[ExpectedType], default: ExpectedType = cast(ExpectedType, sentinel)) -> Optional[ExpectedType]:  # noqa: B008
+    def type_awared_get(self, key: str, hint: Type[ExpectedType], default: Union[ExpectedType, InnerNoneType] = sentinel) -> Optional[ExpectedType]:
         for source in self.sources:
             maybe_result = source.type_awared_get(key, hint, default=default)
             if maybe_result is not default:
                 return maybe_result
 
         if default is not sentinel:
-            return default
+            return cast(ExpectedType, default)
 
         return None

skelet/sources/env.py

@@ -1,7 +1,7 @@
 import os
 import platform
 from functools import cached_property
-from typing import Dict, List, Optional, Type, cast
+from typing import Dict, List, Optional, Type, Union, cast
 
 from denial import InnerNoneType
 from printo import repred
@@ -46,15 +46,15 @@ def data(self) -> Dict[str, str]:
 
         return result
 
-    def type_awared_get(self, key: str, hint: Type[ExpectedType], default: ExpectedType = cast(ExpectedType, sentinel)) -> Optional[ExpectedType]:  # noqa: B008
-        subresult = cast(str, self.get(key, default))
+    def type_awared_get(self, key: str, hint: Type[ExpectedType], default: Union[ExpectedType, InnerNoneType] = sentinel) -> Optional[ExpectedType]:
+        subresult = self.get(key, default)
 
         if subresult is default:
-            if default is not sentinel:
-                return default
-            return None
+            if default is sentinel:
+                return None
+            return cast(ExpectedType, default)
 
-        return from_string(subresult, hint)
+        return from_string(cast(str, subresult), hint)
 
     @classmethod
     def for_library(cls, library_name: str) -> List['EnvSource[ExpectedType]']:

skelet/storage.py

@@ -1,13 +1,14 @@
 from collections import defaultdict
 from threading import Lock
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
 
 from denial import InnerNoneType
 from locklib import ContextLockProtocol
-from printo import descript_data_object
+from printo import describe_data_object
 
 from skelet.sources.abstract import AbstractSource, ExpectedType
 from skelet.sources.collection import SourcesCollection
+from skelet.types import InstanceSourceItem
 
 sentinel = InnerNoneType()
 
@@ -17,8 +18,22 @@ 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 _validate_instance_sources(raw: Optional[Sequence['InstanceSourceItem']]) -> Optional[Sequence['InstanceSourceItem']]:
+        if raw is None:
+            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, *, _sources: Optional[Sequence['InstanceSourceItem']] = None, **kwargs: Any) -> None:
+        self.__instance_sources__ = self._validate_instance_sources(_sources)
 
-    def __init__(self, **kwargs: Any) -> None:
         self.__values__: Dict[str, Any] = {}
         self.__locks__ = {field_name: Lock() for field_name in self.__field_names__}
         deduplicated_fields = set(self.__field_names__)
@@ -132,4 +147,4 @@ def __repr__(self) -> str:
             if getattr(type(self), field_name).secret:
                 secrets[field_name] = '***'
 
-        return descript_data_object(type(self).__name__, (), fields_content, placeholders=secrets)  # type: ignore[arg-type]
+        return describe_data_object(type(self).__name__, (), fields_content, placeholders=secrets)  # type: ignore[arg-type]