130 lines
3.8 KiB
Python
130 lines
3.8 KiB
Python
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)
|