Compare commits

...

6 Commits

Author SHA1 Message Date
John Lancaster
0567766b6c passing thru root reference 2026-02-22 21:47:11 -06:00
John Lancaster
66b4c1274f wrap_sub 2026-02-22 21:44:06 -06:00
John Lancaster
0c3dfde336 types module 2026-02-22 21:06:32 -06:00
John Lancaster
8eea62f403 added suppress option 2026-02-22 20:55:59 -06:00
John Lancaster
05ba036346 better new_path 2026-02-22 20:54:05 -06:00
John Lancaster
2efdf19ee1 set/add event 2026-02-22 19:04:00 -06:00
7 changed files with 110 additions and 88 deletions

View File

@@ -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",
]

View File

@@ -1,20 +1,16 @@
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]
_path: MutableSequence[int] _path: MutableSequence[int]
_root: MutableMapping[T] | MutableSequence[T] | None = None _root: MutableMapping[T] | MutableSequence[T] | None = None
hook: HookFunction hook: HookFunction
@@ -22,37 +18,41 @@ class HookedContainer(ABC, Sized, Iterable[T], Container[T]):
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 # Sequence Methods
# __contains__, __iter__, __reversed__, index, and count # __contains__, __iter__, __reversed__, index, and count
@abstractmethod @abstractmethod
def __getitem__(self, key): ... def __getitem__(self, key): ...
def __len__(self):
return len(self._data)
# MutableSequence Methods # MutableSequence Methods
# append, reverse, extend, pop, remove, and __iadd__ # 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: 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): ...
# 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."""

View File

@@ -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: ...

View File

@@ -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,54 +15,58 @@ 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,
): ):
self.hook = hook
match existing: match existing:
case HookedMapping(_data=seq): case HookedMapping(_data=seq):
self._data = seq self._data = seq
case MutableMapping() as seq: case MutableMapping() as seq:
self._data = seq self._data = seq
case _ as it: case _:
self._data = dict(it) raise TypeError(f"Expected a mapping, got {type(existing)}")
# self._data = dict(it)
self._data = existing self._data = existing
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 []
# MutableMapping Methods def __wrap_sub__(self, other, **kwargs):
return HookedMapping(other, hook=self.hook, root=self._root, **kwargs)
def __iter__(self):
return iter(self._data)
def insert(self, key: T, value: MutableNesting[T]) -> None:
self._data[key] = value
if self.hook:
self.hook(e.AddItemEvent(self.new_path(key), value))
# HookedContainer methods
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:
self.insert(key, {}) return self.__wrap_sub__(self.__setitem__(key, {}), path=self.new_path(key))
return HookedMapping(self._data[key], hook=self.hook, path=self.new_path(key))
value = self._data[key] value = self._data[key]
match value: match value:
case HookedMapping(_data=seq):
return type(self)(seq, hook=self.hook, path=self.new_path(key))
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]) -> None: def __setitem__(
if key not in self._data: self,
self.insert(key, value) key: T,
else: value: MutableNesting[T],
self._data[key] = value *,
if self.hook: suppress_hook: bool = False,
self.hook(e.SetItemEvent(self.new_path(key), value)) ) -> 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:
case HookedMapping(_data=value):
self._data[key] = value
case _:
self._data[key] = value
if self.hook and not suppress_hook:
self.hook(event)
return value
def __delitem__(self, key: T) -> None: def __delitem__(self, key: T) -> None:
item = self._data[key] item = self._data.pop(key)
del self._data[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))

View File

@@ -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))

View File

@@ -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)}")

View 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]]