Compare commits
5 Commits
2efdf19ee1
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0567766b6c | ||
|
|
66b4c1274f | ||
|
|
0c3dfde336 | ||
|
|
8eea62f403 | ||
|
|
05ba036346 |
@@ -1,12 +0,0 @@
|
|||||||
from .container import HookedContainer, HookFunction
|
|
||||||
from .mapping import HookedMapping
|
|
||||||
from .sequence import HookedList
|
|
||||||
from .state import NameSpaceState
|
|
||||||
|
|
||||||
__all__ = [
|
|
||||||
"HookedContainer",
|
|
||||||
"HookFunction",
|
|
||||||
"HookedMapping",
|
|
||||||
"HookedList",
|
|
||||||
"NameSpaceState",
|
|
||||||
]
|
|
||||||
|
|||||||
@@ -1,18 +1,13 @@
|
|||||||
from abc import ABC, abstractmethod
|
from abc import ABC, abstractmethod
|
||||||
from collections.abc import Container, Iterable, MutableMapping, MutableSequence, Sized
|
from collections.abc import Container, Iterable, MutableMapping, MutableSequence, Sized
|
||||||
from copy import copy
|
from typing import TypeVar
|
||||||
from typing import Protocol, TypeVar
|
|
||||||
|
|
||||||
from . import events as e
|
from . import events as e
|
||||||
|
from .events import HookFunction
|
||||||
|
from .types import MutableNesting
|
||||||
|
|
||||||
T = TypeVar("T")
|
T = TypeVar("T")
|
||||||
|
|
||||||
type MutableNesting[T] = T | MutableSequence[T] | MutableMapping[T, MutableNesting[T]]
|
|
||||||
|
|
||||||
|
|
||||||
class HookFunction(Protocol):
|
|
||||||
def __call__(self, event: e.ChangeEvent[T]) -> None: ...
|
|
||||||
|
|
||||||
|
|
||||||
class HookedContainer(ABC, Sized, Iterable[T], Container[T]):
|
class HookedContainer(ABC, Sized, Iterable[T], Container[T]):
|
||||||
_data: MutableNesting[T]
|
_data: MutableNesting[T]
|
||||||
@@ -44,12 +39,12 @@ class HookedContainer(ABC, Sized, Iterable[T], Container[T]):
|
|||||||
def __setitem__(self, s, value):
|
def __setitem__(self, s, value):
|
||||||
self._data[s] = value
|
self._data[s] = value
|
||||||
if self.hook:
|
if self.hook:
|
||||||
self.hook(e.SetItemEvent(self.new_path(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.pop(s)
|
||||||
if self.hook:
|
if self.hook:
|
||||||
self.hook(e.RemoveItemEvent(self.new_path(s), item))
|
self.hook(e.RemoveItemEvent(self._root, self.new_path(s), item))
|
||||||
|
|
||||||
# @abstractmethod
|
# @abstractmethod
|
||||||
# def insert(self, index, value): ...
|
# def insert(self, index, value): ...
|
||||||
@@ -57,9 +52,7 @@ class HookedContainer(ABC, Sized, Iterable[T], Container[T]):
|
|||||||
# Custom Methods
|
# Custom Methods
|
||||||
|
|
||||||
def new_path(self, key):
|
def new_path(self, key):
|
||||||
new_path = copy(self._path)
|
return (*self._path, key)
|
||||||
new_path.append(key)
|
|
||||||
return new_path
|
|
||||||
|
|
||||||
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,26 +1,33 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
from collections.abc import MutableSequence
|
from collections.abc import MutableSequence
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass, field
|
||||||
from typing import Generic, TypeVar
|
from typing import TYPE_CHECKING, Generic, Protocol, TypeVar
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from .mapping import HookedMapping
|
||||||
|
|
||||||
T = TypeVar("T")
|
T = TypeVar("T")
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class ChangeEvent(Generic[T]):
|
class ChangeEvent(Generic[T]):
|
||||||
|
root: HookedMapping[T] = field(repr=False)
|
||||||
path: MutableSequence[int]
|
path: MutableSequence[int]
|
||||||
item: T
|
item: T
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class AddItemEvent(ChangeEvent[T]):
|
class AddItemEvent(ChangeEvent[T]): ...
|
||||||
item: T
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class SetItemEvent(ChangeEvent[T]):
|
class SetItemEvent(ChangeEvent[T]): ...
|
||||||
item: T
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class RemoveItemEvent(ChangeEvent[T]):
|
class RemoveItemEvent(ChangeEvent[T]): ...
|
||||||
item: T
|
|
||||||
|
|
||||||
|
class HookFunction(Protocol):
|
||||||
|
def __call__(self, event: ChangeEvent[T]) -> None: ...
|
||||||
|
|||||||
@@ -2,7 +2,8 @@ from collections.abc import MutableMapping, MutableSequence
|
|||||||
from typing import TypeVar
|
from typing import TypeVar
|
||||||
|
|
||||||
from . import events as e
|
from . import events as e
|
||||||
from .container import HookedContainer, HookFunction, MutableNesting
|
from .container import HookedContainer, MutableNesting
|
||||||
|
from .events import HookFunction
|
||||||
|
|
||||||
T = TypeVar("T")
|
T = TypeVar("T")
|
||||||
|
|
||||||
@@ -14,6 +15,7 @@ class HookedMapping(HookedContainer[T], MutableMapping[T, MutableNesting[T]]):
|
|||||||
self,
|
self,
|
||||||
existing: MutableMapping[T, MutableNesting[T]],
|
existing: MutableMapping[T, MutableNesting[T]],
|
||||||
hook: HookFunction | None = None,
|
hook: HookFunction | None = None,
|
||||||
|
root: MutableMapping[T] | None = None,
|
||||||
path: MutableSequence[int] | None = None,
|
path: MutableSequence[int] | None = None,
|
||||||
):
|
):
|
||||||
match existing:
|
match existing:
|
||||||
@@ -26,31 +28,45 @@ class HookedMapping(HookedContainer[T], MutableMapping[T, MutableNesting[T]]):
|
|||||||
# self._data = dict(it)
|
# self._data = dict(it)
|
||||||
self._data = existing
|
self._data = existing
|
||||||
self.hook = hook
|
self.hook = hook
|
||||||
|
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 []
|
||||||
|
|
||||||
|
def __wrap_sub__(self, other, **kwargs):
|
||||||
|
return HookedMapping(other, hook=self.hook, root=self._root, **kwargs)
|
||||||
|
|
||||||
def __getitem__(self, key: T) -> MutableNesting[T]:
|
def __getitem__(self, key: T) -> MutableNesting[T]:
|
||||||
if key not in self._data:
|
if key not in self._data:
|
||||||
return self.__setitem__(key, {})
|
return self.__wrap_sub__(self.__setitem__(key, {}), path=self.new_path(key))
|
||||||
value = self._data[key]
|
value = self._data[key]
|
||||||
match value:
|
match value:
|
||||||
case MutableMapping() as mapping:
|
case MutableMapping() as mapping:
|
||||||
return HookedMapping(mapping, hook=self.hook, path=self.new_path(key))
|
return self.__wrap_sub__(mapping, path=self.new_path(key))
|
||||||
case _ as item:
|
case _ as item:
|
||||||
return item
|
return item
|
||||||
|
|
||||||
def __setitem__(self, key: T, value: MutableNesting[T]) -> MutableNesting[T] | None:
|
def __setitem__(
|
||||||
|
self,
|
||||||
|
key: T,
|
||||||
|
value: MutableNesting[T],
|
||||||
|
*,
|
||||||
|
suppress_hook: bool = False,
|
||||||
|
) -> MutableNesting[T] | None:
|
||||||
new_path = self.new_path(key)
|
new_path = self.new_path(key)
|
||||||
event = e.SetItemEvent(new_path, value) if key in self._data else e.AddItemEvent(new_path, value)
|
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=value):
|
||||||
self._data[key] = value
|
self._data[key] = value
|
||||||
case _:
|
case _:
|
||||||
self._data[key] = value
|
self._data[key] = value
|
||||||
if self.hook:
|
if self.hook and not suppress_hook:
|
||||||
self.hook(event)
|
self.hook(event)
|
||||||
return value
|
return value
|
||||||
|
|
||||||
def __delitem__(self, key: T) -> None:
|
def __delitem__(self, key: T) -> None:
|
||||||
item = self._data.pop(key)
|
item = self._data.pop(key)
|
||||||
if self.hook:
|
if self.hook:
|
||||||
self.hook(e.RemoveItemEvent(self.new_path(key), item))
|
self.hook(e.RemoveItemEvent(self._root, self.new_path(key), item))
|
||||||
|
|||||||
@@ -3,7 +3,9 @@ from copy import copy
|
|||||||
from typing import TypeVar
|
from typing import TypeVar
|
||||||
|
|
||||||
from . import events as e
|
from . import events as e
|
||||||
from .container import HookedContainer, HookFunction
|
from .container import HookedContainer
|
||||||
|
from .events import HookFunction
|
||||||
|
from .types import MutableNesting
|
||||||
|
|
||||||
T = TypeVar("T")
|
T = TypeVar("T")
|
||||||
|
|
||||||
@@ -16,6 +18,7 @@ class HookedList(HookedContainer[T], MutableSequence[T]):
|
|||||||
self,
|
self,
|
||||||
existing: MutableSequence[T],
|
existing: MutableSequence[T],
|
||||||
hook: HookFunction | None = None,
|
hook: HookFunction | None = None,
|
||||||
|
root: MutableNesting[T] | None = None,
|
||||||
path: MutableSequence[int] | None = None,
|
path: MutableSequence[int] | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
self.hook = hook
|
self.hook = hook
|
||||||
@@ -26,6 +29,7 @@ class HookedList(HookedContainer[T], MutableSequence[T]):
|
|||||||
self._data = seq
|
self._data = seq
|
||||||
case _ as it:
|
case _ as it:
|
||||||
self._data = list(it)
|
self._data = list(it)
|
||||||
|
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 []
|
||||||
|
|
||||||
def __getitem__(self, s):
|
def __getitem__(self, s):
|
||||||
@@ -33,13 +37,13 @@ class HookedList(HookedContainer[T], MutableSequence[T]):
|
|||||||
new_path.append(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, path=new_path)
|
return type(self)(seq, self.hook, root=self._root, path=new_path)
|
||||||
case MutableSequence() as seq:
|
case MutableSequence() as seq:
|
||||||
return HookedList(seq, self.hook, 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):
|
def insert(self, index, value):
|
||||||
self._data.insert(index, value)
|
self._data.insert(index, value)
|
||||||
if self.hook:
|
if self.hook:
|
||||||
self.hook(e.AddItemEvent(self.new_path(index), value))
|
self.hook(e.AddItemEvent(self._root, self.new_path(index), value))
|
||||||
|
|||||||
@@ -1,31 +1,43 @@
|
|||||||
|
from collections.abc import MutableMapping
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
|
||||||
from .mapping import HookedMapping
|
from .mapping import HookedMapping
|
||||||
|
|
||||||
|
|
||||||
class EntityState(HookedMapping[str]):
|
class EntityState(HookedMapping[str]):
|
||||||
pass
|
def __setitem__(self, key, value):
|
||||||
|
super().__setitem__(key, value)
|
||||||
|
super().__setitem__("last_changed", datetime.now(), suppress_hook=True)
|
||||||
|
|
||||||
|
|
||||||
class DomainState(HookedMapping[str]):
|
class DomainState(HookedMapping[str]):
|
||||||
_data: EntityState
|
_data: MutableMapping[str, EntityState]
|
||||||
|
|
||||||
def __setitem__(self, key, value):
|
def __getitem__(self, key):
|
||||||
super().__setitem__(key, value)
|
match super().__getitem__(key):
|
||||||
super().__setitem__("last_changed", datetime.now())
|
case HookedMapping(_data=val):
|
||||||
|
return EntityState(val, hook=self.hook, path=self.new_path(key))
|
||||||
|
case _ as val:
|
||||||
|
raise TypeError(f"Expected a mapping for domain state, got {type(val)}")
|
||||||
|
|
||||||
|
|
||||||
class NameSpaceState(HookedMapping[str]):
|
class NameSpaceState(HookedMapping[str]):
|
||||||
_data: DomainState
|
_data: MutableMapping[str, DomainState]
|
||||||
|
|
||||||
def __iter__(self):
|
|
||||||
return super().__iter__()
|
|
||||||
|
|
||||||
def __setitem__(self, key, value):
|
|
||||||
super().__setitem__(key, value)
|
|
||||||
# print("ns SetItem")
|
|
||||||
|
|
||||||
def __getitem__(self, key):
|
def __getitem__(self, key):
|
||||||
val = super().__getitem__(key)
|
match super().__getitem__(key):
|
||||||
# print("ns GetItem")
|
case HookedMapping(_data=val):
|
||||||
return DomainState(val, hook=self.hook, path=self.new_path(key))
|
return DomainState(val, hook=self.hook, path=self.new_path(key))
|
||||||
|
case _ as val:
|
||||||
|
raise TypeError(f"Expected a mapping for domain state, got {type(val)}")
|
||||||
|
|
||||||
|
|
||||||
|
class FullState(HookedMapping[str]):
|
||||||
|
_data: MutableMapping[str, NameSpaceState]
|
||||||
|
|
||||||
|
def __getitem__(self, key):
|
||||||
|
match super().__getitem__(key):
|
||||||
|
case HookedMapping(_data=val):
|
||||||
|
return NameSpaceState(val, hook=self.hook, path=self.new_path(key))
|
||||||
|
case _ as val:
|
||||||
|
raise TypeError(f"Expected a mapping for namespace state, got {type(val)}")
|
||||||
|
|||||||
6
src/hooked_containers/types.py
Normal file
6
src/hooked_containers/types.py
Normal file
@@ -0,0 +1,6 @@
|
|||||||
|
from collections.abc import MutableMapping, MutableSequence
|
||||||
|
from typing import TypeVar
|
||||||
|
|
||||||
|
T = TypeVar("T")
|
||||||
|
|
||||||
|
type MutableNesting[T] = T | MutableSequence[T] | MutableMapping[T, MutableNesting[T]]
|
||||||
Reference in New Issue
Block a user