diff --git a/src/hooked_containers/mixins.py b/src/hooked_containers/mixins.py new file mode 100644 index 0000000..c564baa --- /dev/null +++ b/src/hooked_containers/mixins.py @@ -0,0 +1,129 @@ +from collections.abc import Collection +from collections.abc import Container +from collections.abc import Iterable +from collections.abc import Iterator +from collections.abc import Mapping +from collections.abc import MutableMapping +from collections.abc import MutableSequence +from collections.abc import Sequence +from typing import Any +from typing import overload + +from .events import AddItemEvent +from .events import ChangeEvent +from .events import HookFunction +from .events import RemoveItemEvent +from .events import SetItemEvent + + +class ContainerMixin[T](Container[T]): + """Provide membership testing against subclass-owned storage.""" + + _data: Container[T] + + def __contains__(self, item: object) -> bool: + return item in self._data + + +class CollectionMixin[T](ContainerMixin[T], Collection[T]): + """Add iteration and size to the container interface.""" + + _data: Collection[T] + + def __iter__(self) -> Iterator[T]: + return iter(self._data) + + def __len__(self) -> int: + return len(self._data) + + +class HookMixin: + """Share optional mutation callbacks and root-relative event paths.""" + + hook: HookFunction | None = None + _root: Any = None + _path: Sequence[Any] = () + + def new_path(self, key: Any) -> list[Any]: + return [*self._path, key] + + def _notify(self, event_type: type[ChangeEvent[Any]], key: Any, item: Any) -> None: + if self.hook is not None: + root = self._root if self._root is not None else self + self.hook(event_type(root, self.new_path(key), item)) + + +class SequenceMixin[T](CollectionMixin[T], Sequence[T]): + """Add indexed access and the standard read-only sequence helpers.""" + + _data: Sequence[T] + + @overload + def __getitem__(self, index: int) -> T: ... + + @overload + def __getitem__(self, index: slice) -> Sequence[T]: ... + + def __getitem__(self, index: int | slice) -> T | Sequence[T]: + return self._data[index] + + +class MutableSequenceMixin[T](HookMixin, SequenceMixin[T], MutableSequence[T]): + """Report primitive mutations used by the mutable sequence helpers.""" + + _data: MutableSequence[T] + + @overload + def __getitem__(self, index: int) -> T: ... + + @overload + def __getitem__(self, index: slice) -> MutableSequence[T]: ... + + def __getitem__(self, index: int | slice) -> T | MutableSequence[T]: + return self._data[index] + + @overload + def __setitem__(self, index: int, value: T) -> None: ... + + @overload + def __setitem__(self, index: slice, value: Iterable[T]) -> None: ... + + def __setitem__(self, index: Any, value: Any) -> None: + if isinstance(index, slice): + value = list(value) + self._data[index] = value + self._notify(SetItemEvent, index, value) + + def __delitem__(self, index: int | slice) -> None: + item = self._data[index] + del self._data[index] + self._notify(RemoveItemEvent, index, item) + + def insert(self, index: int, value: T) -> None: + self._data.insert(index, value) + self._notify(AddItemEvent, index, value) + + +class MappingMixin[K, V](CollectionMixin[K], Mapping[K, V]): + """Add key-based access and the standard read-only mapping helpers.""" + + _data: Mapping[K, V] + + def __getitem__(self, key: K) -> V: + return self._data[key] + + +class MutableMappingMixin[K, V](HookMixin, MappingMixin[K, V], MutableMapping[K, V]): + """Report primitive mutations used by the mutable mapping helpers.""" + + _data: MutableMapping[K, V] + + def __setitem__(self, key: K, value: V) -> None: + event_type = SetItemEvent if key in self._data else AddItemEvent + self._data[key] = value + self._notify(event_type, key, value) + + def __delitem__(self, key: K) -> None: + item = self._data[key] + del self._data[key] + self._notify(RemoveItemEvent, key, item)