Compare commits

...

5 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
7 changed files with 86 additions and 60 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,18 +1,13 @@
from abc import ABC, abstractmethod
from collections.abc import Container, Iterable, MutableMapping, MutableSequence, Sized
from copy import copy
from typing import Protocol, TypeVar
from typing import TypeVar
from . import events as e
from .events import HookFunction
from .types import MutableNesting
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]):
_data: MutableNesting[T]
@@ -44,12 +39,12 @@ class HookedContainer(ABC, Sized, Iterable[T], Container[T]):
def __setitem__(self, s, value):
self._data[s] = value
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):
item = self._data.pop(s)
if self.hook:
self.hook(e.RemoveItemEvent(self.new_path(s), item))
self.hook(e.RemoveItemEvent(self._root, self.new_path(s), item))
# @abstractmethod
# def insert(self, index, value): ...
@@ -57,9 +52,7 @@ class HookedContainer(ABC, Sized, Iterable[T], Container[T]):
# Custom Methods
def new_path(self, key):
new_path = copy(self._path)
new_path.append(key)
return new_path
return (*self._path, key)
def get_nested(self, keys):
"""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 dataclasses import dataclass
from typing import Generic, TypeVar
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Generic, Protocol, TypeVar
if TYPE_CHECKING:
from .mapping import HookedMapping
T = TypeVar("T")
@dataclass(frozen=True)
class ChangeEvent(Generic[T]):
root: HookedMapping[T] = field(repr=False)
path: MutableSequence[int]
item: T
@dataclass(frozen=True)
class AddItemEvent(ChangeEvent[T]):
item: T
class AddItemEvent(ChangeEvent[T]): ...
@dataclass(frozen=True)
class SetItemEvent(ChangeEvent[T]):
item: T
class SetItemEvent(ChangeEvent[T]): ...
@dataclass(frozen=True)
class RemoveItemEvent(ChangeEvent[T]):
item: T
class RemoveItemEvent(ChangeEvent[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 . import events as e
from .container import HookedContainer, HookFunction, MutableNesting
from .container import HookedContainer, MutableNesting
from .events import HookFunction
T = TypeVar("T")
@@ -14,6 +15,7 @@ class HookedMapping(HookedContainer[T], MutableMapping[T, MutableNesting[T]]):
self,
existing: MutableMapping[T, MutableNesting[T]],
hook: HookFunction | None = None,
root: MutableMapping[T] | None = None,
path: MutableSequence[int] | None = None,
):
match existing:
@@ -26,31 +28,45 @@ class HookedMapping(HookedContainer[T], MutableMapping[T, MutableNesting[T]]):
# self._data = dict(it)
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 []
def __wrap_sub__(self, other, **kwargs):
return HookedMapping(other, hook=self.hook, root=self._root, **kwargs)
def __getitem__(self, key: T) -> MutableNesting[T]:
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]
match value:
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:
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)
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:
case HookedMapping(_data=value):
self._data[key] = value
case _:
self._data[key] = value
if self.hook:
if self.hook and not suppress_hook:
self.hook(event)
return value
def __delitem__(self, key: T) -> None:
item = self._data.pop(key)
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 . import events as e
from .container import HookedContainer, HookFunction
from .container import HookedContainer
from .events import HookFunction
from .types import MutableNesting
T = TypeVar("T")
@@ -16,6 +18,7 @@ class HookedList(HookedContainer[T], MutableSequence[T]):
self,
existing: MutableSequence[T],
hook: HookFunction | None = None,
root: MutableNesting[T] | None = None,
path: MutableSequence[int] | None = None,
) -> None:
self.hook = hook
@@ -26,6 +29,7 @@ class HookedList(HookedContainer[T], MutableSequence[T]):
self._data = seq
case _ as 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 []
def __getitem__(self, s):
@@ -33,13 +37,13 @@ class HookedList(HookedContainer[T], MutableSequence[T]):
new_path.append(s)
match self._data[s]:
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:
return HookedList(seq, self.hook, path=new_path)
return HookedList(seq, self.hook, root=self._root, path=new_path)
case _ as item:
return item
def insert(self, index, value):
self._data.insert(index, value)
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 .mapping import HookedMapping
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]):
_data: EntityState
_data: MutableMapping[str, EntityState]
def __setitem__(self, key, value):
super().__setitem__(key, value)
super().__setitem__("last_changed", datetime.now())
def __getitem__(self, key):
match super().__getitem__(key):
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]):
_data: DomainState
def __iter__(self):
return super().__iter__()
def __setitem__(self, key, value):
super().__setitem__(key, value)
# print("ns SetItem")
_data: MutableMapping[str, DomainState]
def __getitem__(self, key):
val = super().__getitem__(key)
# print("ns GetItem")
return DomainState(val, hook=self.hook, path=self.new_path(key))
match super().__getitem__(key):
case HookedMapping(_data=val):
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]]