typing updates
This commit is contained in:
@@ -1,58 +1,31 @@
|
|||||||
from abc import ABC, abstractmethod
|
from abc import ABC
|
||||||
from collections.abc import Container, Iterable, MutableMapping, MutableSequence, Sized
|
from abc import abstractmethod
|
||||||
from typing import TypeVar
|
from collections.abc import MutableMapping
|
||||||
|
from collections.abc import MutableSequence
|
||||||
|
|
||||||
from . import events as e
|
from . import events as e
|
||||||
from .events import HookFunction
|
from .mixins import CollectionMixin
|
||||||
|
from .mixins import HookMixin
|
||||||
from .types import MutableNesting
|
from .types import MutableNesting
|
||||||
|
|
||||||
T = TypeVar("T")
|
|
||||||
|
|
||||||
|
class HookedContainer[T](HookMixin, CollectionMixin[T], ABC):
|
||||||
class HookedContainer(ABC, Sized, Iterable[T], Container[T]):
|
_data: MutableMapping[T, MutableNesting[T]] | MutableSequence[T]
|
||||||
_data: MutableNesting[T]
|
|
||||||
_path: MutableSequence[int]
|
|
||||||
_root: MutableMapping[T] | MutableSequence[T] | None = None
|
|
||||||
hook: HookFunction
|
|
||||||
|
|
||||||
def __repr__(self):
|
def __repr__(self):
|
||||||
return f"{self.__class__.__name__}({self._data!r})"
|
return f"{self.__class__.__name__}({self._data!r})"
|
||||||
|
|
||||||
def __len__(self):
|
|
||||||
return len(self._data)
|
|
||||||
|
|
||||||
def __iter__(self):
|
|
||||||
return iter(self._data)
|
|
||||||
|
|
||||||
def __contains__(self, x):
|
|
||||||
return x in self._data
|
|
||||||
|
|
||||||
# Sequence Methods
|
|
||||||
# __contains__, __iter__, __reversed__, index, and count
|
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def __getitem__(self, key): ...
|
def __getitem__(self, key): ...
|
||||||
|
|
||||||
# MutableSequence Methods
|
|
||||||
# append, reverse, extend, pop, remove, and __iadd__
|
|
||||||
|
|
||||||
def __setitem__(self, s, value):
|
def __setitem__(self, s, value):
|
||||||
self._data[s] = value
|
self._data[s] = value
|
||||||
if self.hook:
|
self._notify(e.SetItemEvent, s, value)
|
||||||
self.hook(e.SetItemEvent(self._root, self.new_path(s), value))
|
|
||||||
|
|
||||||
def __delitem__(self, s):
|
def __delitem__(self, s):
|
||||||
item = self._data.pop(s)
|
item = self._data[s]
|
||||||
if self.hook:
|
del self._data[s]
|
||||||
self.hook(e.RemoveItemEvent(self._root, self.new_path(s), item))
|
self._notify(e.RemoveItemEvent, s, item)
|
||||||
|
|
||||||
# @abstractmethod
|
|
||||||
# def insert(self, index, value): ...
|
|
||||||
|
|
||||||
# Custom Methods
|
|
||||||
|
|
||||||
def new_path(self, key):
|
|
||||||
return (*self._path, key)
|
|
||||||
|
|
||||||
def get_nested(self, keys):
|
def get_nested(self, keys):
|
||||||
"""Recursively call __getitem__ with each key in the iterable."""
|
"""Recursively call __getitem__ with each key in the iterable."""
|
||||||
|
|||||||
@@ -1,33 +1,33 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from collections.abc import MutableSequence
|
from collections.abc import Collection
|
||||||
from dataclasses import dataclass, field
|
from collections.abc import Hashable
|
||||||
from typing import TYPE_CHECKING, Generic, Protocol, TypeVar
|
from collections.abc import Sequence
|
||||||
|
from dataclasses import dataclass
|
||||||
if TYPE_CHECKING:
|
from dataclasses import field
|
||||||
from .mapping import HookedMapping
|
from typing import Protocol
|
||||||
|
|
||||||
T = TypeVar("T")
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class ChangeEvent(Generic[T]):
|
class ChangeEvent[T]:
|
||||||
root: HookedMapping[T] = field(repr=False)
|
"""Describe a mutation at an index or key path within a root collection."""
|
||||||
path: MutableSequence[int]
|
|
||||||
|
root: Collection[object] = field(repr=False)
|
||||||
|
path: Sequence[Hashable | slice]
|
||||||
item: T
|
item: T
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class AddItemEvent(ChangeEvent[T]): ...
|
class AddItemEvent[T](ChangeEvent[T]): ...
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class SetItemEvent(ChangeEvent[T]): ...
|
class SetItemEvent[T](ChangeEvent[T]): ...
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class RemoveItemEvent(ChangeEvent[T]): ...
|
class RemoveItemEvent[T](ChangeEvent[T]): ...
|
||||||
|
|
||||||
|
|
||||||
class HookFunction(Protocol):
|
class HookFunction(Protocol):
|
||||||
def __call__(self, event: ChangeEvent[T]) -> None: ...
|
def __call__[T](self, event: ChangeEvent[T]) -> None: ...
|
||||||
|
|||||||
@@ -1,14 +1,14 @@
|
|||||||
from collections.abc import MutableMapping, MutableSequence
|
from collections.abc import MutableMapping
|
||||||
from typing import TypeVar
|
from collections.abc import MutableSequence
|
||||||
|
|
||||||
from . import events as e
|
from . import events as e
|
||||||
from .container import HookedContainer, MutableNesting
|
from .container import HookedContainer
|
||||||
|
from .container import MutableNesting
|
||||||
from .events import HookFunction
|
from .events import HookFunction
|
||||||
|
from .mixins import MutableMappingMixin
|
||||||
T = TypeVar("T")
|
|
||||||
|
|
||||||
|
|
||||||
class HookedMapping(HookedContainer[T], MutableMapping[T, MutableNesting[T]]):
|
class HookedMapping[T](MutableMappingMixin[T, MutableNesting[T]], HookedContainer[T]):
|
||||||
_data: MutableMapping[T, MutableNesting[T]]
|
_data: MutableMapping[T, MutableNesting[T]]
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
@@ -26,7 +26,6 @@ class HookedMapping(HookedContainer[T], MutableMapping[T, MutableNesting[T]]):
|
|||||||
case _:
|
case _:
|
||||||
raise TypeError(f"Expected a mapping, got {type(existing)}")
|
raise TypeError(f"Expected a mapping, got {type(existing)}")
|
||||||
# self._data = dict(it)
|
# self._data = dict(it)
|
||||||
self._data = existing
|
|
||||||
self.hook = hook
|
self.hook = hook
|
||||||
self._root = root if root is not None else self
|
self._root = root if root is not None else self
|
||||||
self._path = list(path) if path is not None else []
|
self._path = list(path) if path is not None else []
|
||||||
@@ -51,22 +50,16 @@ class HookedMapping(HookedContainer[T], MutableMapping[T, MutableNesting[T]]):
|
|||||||
*,
|
*,
|
||||||
suppress_hook: bool = False,
|
suppress_hook: bool = False,
|
||||||
) -> MutableNesting[T] | None:
|
) -> MutableNesting[T] | None:
|
||||||
new_path = self.new_path(key)
|
|
||||||
event = (
|
|
||||||
e.SetItemEvent(self._root, new_path, value)
|
|
||||||
if key in self._data
|
|
||||||
else e.AddItemEvent(self._root, new_path, value)
|
|
||||||
)
|
|
||||||
match value:
|
match value:
|
||||||
case HookedMapping(_data=value):
|
case HookedMapping(_data=data):
|
||||||
self._data[key] = value
|
event_type = e.SetItemEvent if key in self._data else e.AddItemEvent
|
||||||
|
self._data[key] = data
|
||||||
|
if not suppress_hook:
|
||||||
|
self._notify(event_type, key, value)
|
||||||
|
return data
|
||||||
case _:
|
case _:
|
||||||
|
if suppress_hook:
|
||||||
self._data[key] = value
|
self._data[key] = value
|
||||||
if self.hook and not suppress_hook:
|
else:
|
||||||
self.hook(event)
|
super().__setitem__(key, value)
|
||||||
return value
|
return value
|
||||||
|
|
||||||
def __delitem__(self, key: T) -> None:
|
|
||||||
item = self._data.pop(key)
|
|
||||||
if self.hook:
|
|
||||||
self.hook(e.RemoveItemEvent(self._root, self.new_path(key), item))
|
|
||||||
|
|||||||
@@ -1,16 +1,12 @@
|
|||||||
from collections.abc import MutableSequence
|
from collections.abc import MutableSequence
|
||||||
from copy import copy
|
|
||||||
from typing import TypeVar
|
|
||||||
|
|
||||||
from . import events as e
|
|
||||||
from .container import HookedContainer
|
from .container import HookedContainer
|
||||||
from .events import HookFunction
|
from .events import HookFunction
|
||||||
|
from .mixins import MutableSequenceMixin
|
||||||
from .types import MutableNesting
|
from .types import MutableNesting
|
||||||
|
|
||||||
T = TypeVar("T")
|
|
||||||
|
|
||||||
|
class HookedList[T](MutableSequenceMixin[T], HookedContainer[T]):
|
||||||
class HookedList(HookedContainer[T], MutableSequence[T]):
|
|
||||||
_data: MutableSequence[T]
|
_data: MutableSequence[T]
|
||||||
path: MutableSequence[int]
|
path: MutableSequence[int]
|
||||||
|
|
||||||
@@ -33,8 +29,7 @@ class HookedList(HookedContainer[T], MutableSequence[T]):
|
|||||||
self._path = list(path) if path is not None else []
|
self._path = list(path) if path is not None else []
|
||||||
|
|
||||||
def __getitem__(self, s):
|
def __getitem__(self, s):
|
||||||
new_path = copy(self._path)
|
new_path = self.new_path(s)
|
||||||
new_path.append(s)
|
|
||||||
match self._data[s]:
|
match self._data[s]:
|
||||||
case HookedContainer(_data=seq):
|
case HookedContainer(_data=seq):
|
||||||
return type(self)(seq, self.hook, root=self._root, path=new_path)
|
return type(self)(seq, self.hook, root=self._root, path=new_path)
|
||||||
@@ -42,8 +37,3 @@ class HookedList(HookedContainer[T], MutableSequence[T]):
|
|||||||
return HookedList(seq, self.hook, root=self._root, path=new_path)
|
return HookedList(seq, self.hook, root=self._root, path=new_path)
|
||||||
case _ as item:
|
case _ as item:
|
||||||
return item
|
return item
|
||||||
|
|
||||||
def insert(self, index, value):
|
|
||||||
self._data.insert(index, value)
|
|
||||||
if self.hook:
|
|
||||||
self.hook(e.AddItemEvent(self._root, self.new_path(index), value))
|
|
||||||
|
|||||||
Reference in New Issue
Block a user