Compare commits

...
1 Commits
Author SHA1 Message Date
John Lancaster 23d3c113d8 restructured mixins 2026-10-10 09:44:47 -05:00
15 changed files with 2127 additions and 379 deletions
+52 -25
View File
@@ -1,49 +1,76 @@
# Hooked Containers # Hooked Containers
Hooked Containers provides an abstract container foundation that calls a function hook whenever its contents are changed through its API. The aim is to retain familiar Python container operations while letting application code react to additions, replacements, and removals. Hooked Containers adds optional, shallow mutation callbacks to familiar Python container operations. It provides `HookedMapping[K, V]`, `HookedList[T]`, and `HookedSet[T]`, backed by owned `dict`, `list`, and `set` storage. Requires Python 3.12 or newer.
## Mixin Design ## Mixin Design
The design follows Python's [collections.abc abstract base classes](https://docs.python.org/3/library/collections.abc.html#collections-abstract-base-classes). Separate mixin classes should build up functionality in layers, so a custom container can compose the behavior it needs: All reusable container behavior lives in `src/hooked_containers/mixins.py`. The design uses Python's [collections.abc abstract base classes](https://docs.python.org/3/library/collections.abc.html#collections-abstract-base-classes) directly. Three mutable mixins compose shared reads and hook dispatch with their required primitives; standard ABC methods supply the helpers:
| Layer | Responsibility | | Mixin | Backing Storage | Responsibility |
| ---------------- | ------------------------------------------------------------------------------------------- | | --- | --- | --- |
| Container | Membership testing through `__contains__`. | | `ContainerMixin` | Any collection | Containment, iteration, length, and representation. |
| Collection | Iteration and size through `__iter__` and `__len__`, alongside membership. | | `HookMixin` | None | Callback dispatch after successful writes. |
| Hook support | A shared callback and change-event reporting for mutation methods. | | `ListMixin` | `list` | Indexed reads and writes, insertion, and native list operations. |
| Sequence | Indexed access through `__getitem__`, with sequence helpers such as `index` and `count`. | | `MutableMappingMixin` | `dict` | Key-based reads and writes, copying, and merging. |
| Mutable sequence | Assignment, deletion, and insertion, supporting helpers such as `append`, `extend`, and `pop`. | | `MutableSetMixin` | `set` | Addition, discard, and named set operations. |
| Mapping | Key-based access, iteration over keys, and helpers such as `keys`, `items`, and `values`. |
| Mutable mapping | Assignment and deletion, supporting helpers such as `update`, `setdefault`, and `pop`. |
Sequences and mappings are separate branches built on the common container foundation, not successive steps in a single inheritance chain. Mutation hooks belong in the primitive operations so inherited ABC mixin methods also report changes. A compound operation may therefore invoke the hook more than once. `HookMixin` shares callback dispatch across the three mixins. Inherited methods such as `append`, `extend`, `reverse`, `update`, `clear`, and set in-place operators reach the primitive writes automatically. A compound operation may invoke the hook more than once.
This layered separation is the intended architecture. Currently, `HookedContainer` combines shared container and hook behavior; `HookedList` and `HookedMapping` extend it with the standard library's `MutableSequence` and `MutableMapping` mixins, respectively. Each subclass initializes a single `_storage` attribute with the corresponding native container. There is no storage property or separate container-specific read-only hierarchy. Each of `sequence.py`, `mapping.py`, and `set.py` defines just one concrete wrapper, composing its mutable mixin and adding construction. Representation comes from `ContainerMixin`. Dict-style `popitem` intentionally removes the most recently inserted key, unlike the ABC's arbitrary-item implementation.
Custom containers can use the mixins without any concrete wrapper:
```python
from hooked_containers.mixins import MutableMappingMixin
class Settings(MutableMappingMixin[str, int]):
def __init__(self, hook=None):
self._storage = {}
self.hook = hook
```
## Change Hooks ## Change Hooks
A hook receives one change-event object after a mutation. The event identifies the root container (`root`), the index or key path (`path`), and the affected value (`item`). The current event types are `AddItemEvent`, `SetItemEvent`, and `RemoveItemEvent`. A hook receives one immutable `Change(operation, key, value)` record after a write:
| Field | Meaning |
| --- | --- |
| `operation` | `"add"`, `"set"`, or `"remove"` |
| `key` | Mapping key, normalized list index, list slice, or `None` for sets |
| `value` | Added/new value, or the value that was removed |
```python ```python
from hooked_containers.mapping import HookedMapping from hooked_containers import HookedList, HookedMapping, HookedSet
from hooked_containers.sequence import HookedList
def on_change(event): def on_change(change):
print(type(event).__name__, event.path, event.item) print(change.operation, change.key, change.value)
items = HookedList(["first"], hook=on_change) items = HookedList(["first"], hook=on_change)
items.append("second") # AddItemEvent (1,) second items.append("second") # add 1 second
items[0] = "updated" # SetItemEvent (0,) updated items[0] = "updated" # set 0 updated
del items[1] # RemoveItemEvent (1,) second del items[1] # remove 1 second
settings = HookedMapping({}, hook=on_change) settings = HookedMapping({}, hook=on_change)
settings["enabled"] = True # AddItemEvent ('enabled',) True settings["enabled"] = True # add enabled True
members = HookedSet(hook=on_change)
members.add("ready") # add None ready
members.discard("ready") # remove None ready
``` ```
The wrappers can share existing mutable storage. Mutating that storage directly bypasses the hook; make changes through the hooked container to receive events. Hooks are optional, and nested wrappers retain the root and path for supported nested container types. Reads never change data or invoke the callback. Missing mapping keys raise `KeyError`, and `get` respects its default. Integer list keys are normalized against the length before the write; insertions report their clamped destination. Slice keys preserve the supplied slice in pre-write coordinates, and their values are shallow lists of new or removed items. Sorting and in-place repetition report a whole-list `"set"` using `slice(None)`.
## Status Assignments notify even when assigning an equal value. Duplicate set additions, missing set discards, empty clears, and empty slice writes do not notify. Failed primitive writes do not notify; writes completed before a compound operation fails remain reported. A failed sort reports any resulting rearrangement. Callback exceptions propagate after the write without rolling it back. Callbacks that mutate the same container can trigger further callbacks; there is no batching or reentrancy guard.
Requires Python 3.12 or newer. This repository is an early implementation of the container and mixin design; the API and separation of responsibilities may evolve. Construction copies the outer storage without notifying. Nested values are not copied or wrapped, and mutations inside those values are not observed. Copies, slices, and non-mutating derived results are ordinary native containers without hooks. Event values are shallow references, not deep snapshots; later changes to a mutable value remain visible through that reference. Direct changes to private storage bypass callbacks.
Set comparison, algebra, and in-place methods follow the standard ABC implementations. In particular, ABC set operators accept iterables more broadly than built-in `set` operators. These wrappers implement container interfaces, not built-in subclass identity or exact built-in performance guarantees.
## Migration
Custom subclasses now initialize `_storage` directly instead of implementing `_data`. The earlier container/collection hierarchy, read-only sequence, mapping, and set mixins, along with `MutableSequenceMixin`, have been replaced by the shared `ContainerMixin` and the three mutable mixins above. Import reusable mixins from `hooked_containers.mixins` and use native backing storage.
The shallow API replaces the earlier root/path event subclasses and recursive wrapping. Use `change.operation`, `change.key`, and `change.value` instead of event types, `path`, and `item`. `HookedMapping` has independent key and value type parameters. Shared outer storage, autovivification, `suppress_hook`, `get_nested`, and the domain-state classes are no longer supported. The development notebooks are historical experiments with the previous API; the examples above describe the current API.
+2
View File
@@ -16,6 +16,8 @@ build-backend = "uv_build"
[dependency-groups] [dependency-groups]
dev = [ dev = [
"ipykernel>=7.2.0", "ipykernel>=7.2.0",
"ipywidgets>=8.1.9",
"notebook>=7.6.3",
"rich>=14.3.3", "rich>=14.3.3",
"ruff>=0.15.2", "ruff>=0.15.2",
"ty>=0.0.85", "ty>=0.0.85",
+7
View File
@@ -0,0 +1,7 @@
from .events import Change as Change
from .events import HookFunction as HookFunction
from .mapping import HookedMapping as HookedMapping
from .sequence import HookedList as HookedList
from .set import HookedSet as HookedSet
__all__ = ["Change", "HookFunction", "HookedMapping", "HookedList", "HookedSet"]
-35
View File
@@ -1,35 +0,0 @@
from abc import ABC
from abc import abstractmethod
from collections.abc import MutableMapping
from collections.abc import MutableSequence
from . import events as e
from .mixins import CollectionMixin
from .mixins import HookMixin
from .types import MutableNesting
class HookedContainer[T](HookMixin, CollectionMixin[T], ABC):
_data: MutableMapping[T, MutableNesting[T]] | MutableSequence[T]
def __repr__(self):
return f"{self.__class__.__name__}({self._data!r})"
@abstractmethod
def __getitem__(self, key): ...
def __setitem__(self, s, value):
self._data[s] = value
self._notify(e.SetItemEvent, s, value)
def __delitem__(self, s):
item = self._data[s]
del self._data[s]
self._notify(e.RemoveItemEvent, s, item)
def get_nested(self, keys):
"""Recursively call __getitem__ with each key in the iterable."""
result = self
for key in keys:
result = result[key]
return result
+9 -23
View File
@@ -1,33 +1,19 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Collection from collections.abc import Callable
from collections.abc import Hashable from collections.abc import Hashable
from collections.abc import Sequence
from dataclasses import dataclass from dataclasses import dataclass
from dataclasses import field from typing import Any
from typing import Protocol from typing import Literal
@dataclass(frozen=True) @dataclass(frozen=True)
class ChangeEvent[T]: class Change[T]:
"""Describe a mutation at an index or key path within a root collection.""" """Describe a shallow write; removals carry the removed value."""
root: Collection[object] = field(repr=False) operation: Literal["add", "set", "remove"]
path: Sequence[Hashable | slice] key: Hashable | slice | None
item: T value: T
@dataclass(frozen=True) type HookFunction = Callable[[Change[Any]], None]
class AddItemEvent[T](ChangeEvent[T]): ...
@dataclass(frozen=True)
class SetItemEvent[T](ChangeEvent[T]): ...
@dataclass(frozen=True)
class RemoveItemEvent[T](ChangeEvent[T]): ...
class HookFunction(Protocol):
def __call__[T](self, event: ChangeEvent[T]) -> None: ...
+9 -55
View File
@@ -1,65 +1,19 @@
from collections.abc import MutableMapping from collections.abc import Iterable
from collections.abc import MutableSequence from collections.abc import Mapping
from typing import Any
from . import events as e
from .container import HookedContainer
from .container import MutableNesting
from .events import HookFunction from .events import HookFunction
from .mixins import MutableMappingMixin from .mixins import MutableMappingMixin
class HookedMapping[T](MutableMappingMixin[T, MutableNesting[T]], HookedContainer[T]): class HookedMapping[K, V](MutableMappingMixin[K, V]):
_data: MutableMapping[T, MutableNesting[T]] """A shallow mapping with optional mutation notifications."""
def __init__( def __init__(
self, self,
existing: MutableMapping[T, MutableNesting[T]], existing: Mapping[K, V] | Iterable[tuple[K, V]] = (),
hook: HookFunction | None = None, hook: HookFunction | None = None,
root: MutableMapping[T] | None = None, **kwargs: Any,
path: MutableSequence[int] | None = None, ) -> None:
): self._storage = dict(existing, **kwargs)
match existing:
case HookedMapping(_data=seq):
self._data = seq
case MutableMapping() as seq:
self._data = seq
case _:
raise TypeError(f"Expected a mapping, got {type(existing)}")
# self._data = dict(it)
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 []
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.__wrap_sub__(self.__setitem__(key, {}), path=self.new_path(key))
value = self._data[key]
match value:
case MutableMapping() as mapping:
return self.__wrap_sub__(mapping, path=self.new_path(key))
case _ as item:
return item
def __setitem__(
self,
key: T,
value: MutableNesting[T],
*,
suppress_hook: bool = False,
) -> MutableNesting[T] | None:
match value:
case HookedMapping(_data=data):
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 _:
if suppress_hook:
self._data[key] = value
else:
super().__setitem__(key, value)
return value
+207 -82
View File
@@ -1,89 +1,65 @@
from collections.abc import Collection from collections.abc import Collection
from collections.abc import Container
from collections.abc import Iterable from collections.abc import Iterable
from collections.abc import Iterator from collections.abc import Iterator
from collections.abc import Mapping
from collections.abc import MutableMapping from collections.abc import MutableMapping
from collections.abc import MutableSequence from collections.abc import MutableSequence
from collections.abc import Sequence from collections.abc import MutableSet
from operator import index as to_index
from typing import Any from typing import Any
from typing import Literal
from typing import Self
from typing import SupportsIndex
from typing import cast
from typing import overload from typing import overload
from .events import AddItemEvent from .events import Change
from .events import ChangeEvent
from .events import HookFunction 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: class HookMixin:
"""Share optional mutation callbacks and root-relative event paths.""" """Dispatch shallow changes after successful writes."""
hook: HookFunction | None = None hook: HookFunction | None = None
_root: Any = None
_path: Sequence[Any] = ()
def new_path(self, key: Any) -> list[Any]: def _notify(self, operation: Literal["add", "set", "remove"], key: Any, value: Any) -> None:
return [*self._path, key]
def _notify(self, event_type: type[ChangeEvent[Any]], key: Any, item: Any) -> None:
if self.hook is not None: if self.hook is not None:
root = self._root if self._root is not None else self self.hook(Change(operation, key, value))
self.hook(event_type(root, self.new_path(key), item))
class SequenceMixin[T](CollectionMixin[T], Sequence[T]): class ContainerMixin[T](Collection[T]):
"""Add indexed access and the standard read-only sequence helpers.""" """Provide common reads and representation for subclass-owned storage."""
_data: Sequence[T] _storage: Collection[T]
def __contains__(self, item: object) -> bool:
return item in self._storage
def __iter__(self) -> Iterator[T]:
return iter(self._storage)
def __len__(self) -> int:
return len(self._storage)
def __repr__(self) -> str:
return f"{type(self).__name__}({self._storage!r})"
class ListMixin[T](HookMixin, ContainerMixin[T], MutableSequence[T]):
"""Provide list operations and report writes to subclass-owned storage."""
_storage: list[T]
@overload @overload
def __getitem__(self, index: int) -> T: ... def __getitem__(self, index: SupportsIndex) -> T: ...
@overload @overload
def __getitem__(self, index: slice) -> Sequence[T]: ... def __getitem__(self, index: slice) -> list[T]: ...
def __getitem__(self, index: int | slice) -> T | Sequence[T]: def __getitem__(self, index: SupportsIndex | slice) -> T | list[T]:
return self._data[index] return self._storage[index if isinstance(index, slice) else to_index(index)]
class MutableSequenceMixin[T](HookMixin, SequenceMixin[T], MutableSequence[T]):
"""Report primitive mutations used by the mutable sequence helpers."""
_data: MutableSequence[T]
@overload @overload
def __getitem__(self, index: int) -> T: ... def __setitem__(self, index: SupportsIndex, value: T) -> None: ...
@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 @overload
def __setitem__(self, index: slice, value: Iterable[T]) -> None: ... def __setitem__(self, index: slice, value: Iterable[T]) -> None: ...
@@ -91,39 +67,188 @@ class MutableSequenceMixin[T](HookMixin, SequenceMixin[T], MutableSequence[T]):
def __setitem__(self, index: Any, value: Any) -> None: def __setitem__(self, index: Any, value: Any) -> None:
if isinstance(index, slice): if isinstance(index, slice):
value = list(value) value = list(value)
self._data[index] = value previous = self._storage[index]
self._notify(SetItemEvent, index, value) self._storage[index] = value
if previous or value:
self._notify("set", index, value)
return
index = to_index(index)
actual_index = index if index >= 0 else len(self) + index
self._storage[index] = value
self._notify("set", actual_index, value)
def __delitem__(self, index: int | slice) -> None: def __delitem__(self, index: SupportsIndex | slice) -> None:
item = self._data[index] if not isinstance(index, slice):
del self._data[index] index = to_index(index)
self._notify(RemoveItemEvent, index, item) item = self._storage[index]
actual_index = index if isinstance(index, slice) or index >= 0 else len(self) + index
del self._storage[index]
if not isinstance(index, slice) or item:
self._notify("remove", actual_index, item)
def insert(self, index: int, value: T) -> None: def insert(self, index: SupportsIndex, value: T) -> None:
self._data.insert(index, value) index = to_index(index)
self._notify(AddItemEvent, index, value) index = min(index, len(self)) if index >= 0 else max(0, len(self) + index)
self._storage.insert(index, value)
self._notify("add", index, value)
@staticmethod
def _operand(other: Any) -> Any:
return other._storage if isinstance(other, ListMixin) else other
def __eq__(self, other: object) -> bool:
return self._storage.__eq__(self._operand(other))
def __lt__(self, other: Any) -> bool:
return self._storage.__lt__(self._operand(other))
def __le__(self, other: Any) -> bool:
return self._storage.__le__(self._operand(other))
def __gt__(self, other: Any) -> bool:
return self._storage.__gt__(self._operand(other))
def __ge__(self, other: Any) -> bool:
return self._storage.__ge__(self._operand(other))
def copy(self) -> list[T]:
return self._storage.copy()
def __add__(self, other: Any) -> list[T]:
return self._storage.__add__(self._operand(other))
def __radd__(self, other: Any) -> list[T]:
if not isinstance(other, list):
return NotImplemented
return other + self._storage
def __mul__(self, count: SupportsIndex) -> list[T]:
return self._storage * count
def __rmul__(self, count: SupportsIndex) -> list[T]:
return self * count
def __imul__(self, count: SupportsIndex) -> Self:
count = to_index(count)
if self and count != 1:
self[:] = self._storage * count
return self
def sort(self, *, key: Any = None, reverse: bool = False) -> None:
values = self._storage
before = tuple(map(id, values))
try:
values.sort(key=key, reverse=reverse)
except Exception:
if tuple(map(id, values)) != before:
self._notify("set", slice(None), list(values))
raise
if values:
self._notify("set", slice(None), list(values))
class MappingMixin[K, V](CollectionMixin[K], Mapping[K, V]): class MutableMappingMixin[K, V](HookMixin, ContainerMixin[K], MutableMapping[K, V]):
"""Add key-based access and the standard read-only mapping helpers.""" """Provide mapping operations and report writes to subclass-owned storage."""
_data: Mapping[K, V] _storage: dict[K, V]
def __getitem__(self, key: K) -> V: def __getitem__(self, key: K) -> V:
return self._data[key] return self._storage[key]
def __reversed__(self) -> Iterator[K]:
return reversed(self._storage)
class MutableMappingMixin[K, V](HookMixin, MappingMixin[K, V], MutableMapping[K, V]): def copy(self) -> dict[K, V]:
"""Report primitive mutations used by the mutable mapping helpers.""" return self._storage.copy()
_data: MutableMapping[K, V] def __or__(self, other: Any) -> dict[K, V]:
if not isinstance(other, (dict, MutableMappingMixin)):
return NotImplemented
return self._storage | dict(other)
def __ror__(self, other: Any) -> dict[K, V]:
if not isinstance(other, (dict, MutableMappingMixin)):
return NotImplemented
return dict(other) | self._storage
def __setitem__(self, key: K, value: V) -> None: def __setitem__(self, key: K, value: V) -> None:
event_type = SetItemEvent if key in self._data else AddItemEvent operation: Literal["add", "set"] = "set" if key in self._storage else "add"
self._data[key] = value self._storage[key] = value
self._notify(event_type, key, value) self._notify(operation, key, value)
def __delitem__(self, key: K) -> None: def __delitem__(self, key: K) -> None:
item = self._data[key] item = self._storage[key]
del self._data[key] del self._storage[key]
self._notify(RemoveItemEvent, key, item) self._notify("remove", key, item)
def popitem(self) -> tuple[K, V]:
if not self:
raise KeyError("popitem(): mapping is empty")
key = next(reversed(self))
value = self[key]
del self[key]
return key, value
def __ior__(self, other: Any) -> Self:
self.update(other)
return self
class MutableSetMixin[T](HookMixin, ContainerMixin[T], MutableSet[T]):
"""Provide set operations and report writes to subclass-owned storage."""
_storage: set[T]
@classmethod
def _from_iterable[S](cls, iterable: Iterable[S]) -> set[S]:
return set(iterable)
def copy(self) -> set[T]:
return self._storage.copy()
def union(self, *others: Iterable[T]) -> set[T]:
return self._storage.union(*others)
def intersection(self, *others: Iterable[T]) -> set[T]:
return self._storage.intersection(*others)
def difference(self, *others: Iterable[T]) -> set[T]:
return self._storage.difference(*others)
def symmetric_difference(self, other: Iterable[T]) -> set[T]:
return self._storage.symmetric_difference(other)
def issubset(self, other: Iterable[T]) -> bool:
return self._storage.issubset(other)
def issuperset(self, other: Iterable[T]) -> bool:
return self._storage.issuperset(other)
def add(self, value: T) -> None:
size = len(self._storage)
self._storage.add(value)
if len(self._storage) != size:
self._notify("add", None, value)
def discard(self, value: T) -> None:
if value in self._storage:
removed = value
if self.hook is not None:
removed = next(item for item in self._storage if item is value or item == value)
self._storage.remove(value)
self._notify("remove", None, removed)
def update(self, *others: Iterable[T]) -> None:
for other in others:
self.__ior__(cast(Any, other))
def difference_update(self, *others: Iterable[T]) -> None:
for other in others:
self.__isub__(cast(Any, other))
def intersection_update(self, *others: Iterable[T]) -> None:
remaining = self.intersection(*others)
self.__iand__(remaining)
def symmetric_difference_update(self, other: Iterable[T]) -> None:
self.__ixor__(cast(Any, other))
+6 -33
View File
@@ -1,39 +1,12 @@
from collections.abc import MutableSequence from collections.abc import Iterable
from .container import HookedContainer
from .events import HookFunction from .events import HookFunction
from .mixins import MutableSequenceMixin from .mixins import ListMixin
from .types import MutableNesting
class HookedList[T](MutableSequenceMixin[T], HookedContainer[T]): class HookedList[T](ListMixin[T]):
_data: MutableSequence[T] """A shallow list with optional mutation notifications."""
path: MutableSequence[int]
def __init__( def __init__(self, existing: Iterable[T] = (), hook: HookFunction | None = None) -> None:
self, self._storage = list(existing)
existing: MutableSequence[T],
hook: HookFunction | None = None,
root: MutableNesting[T] | None = None,
path: MutableSequence[int] | None = None,
) -> None:
self.hook = hook self.hook = hook
match existing:
case HookedContainer(_data=seq):
self._data = seq
case MutableSequence() as seq:
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):
new_path = self.new_path(s)
match self._data[s]:
case HookedContainer(_data=seq):
return type(self)(seq, self.hook, root=self._root, path=new_path)
case MutableSequence() as seq:
return HookedList(seq, self.hook, root=self._root, path=new_path)
case _ as item:
return item
+12
View File
@@ -0,0 +1,12 @@
from collections.abc import Iterable
from .events import HookFunction
from .mixins import MutableSetMixin
class HookedSet[T](MutableSetMixin[T]):
"""A shallow set with optional mutation notifications."""
def __init__(self, existing: Iterable[T] = (), hook: HookFunction | None = None) -> None:
self._storage = set(existing)
self.hook = hook
-43
View File
@@ -1,43 +0,0 @@
from collections.abc import MutableMapping
from datetime import datetime
from .mapping import HookedMapping
class EntityState(HookedMapping[str]):
def __setitem__(self, key, value):
super().__setitem__(key, value)
super().__setitem__("last_changed", datetime.now(), suppress_hook=True)
class DomainState(HookedMapping[str]):
_data: MutableMapping[str, EntityState]
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: MutableMapping[str, DomainState]
def __getitem__(self, 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)}")
-6
View File
@@ -1,6 +0,0 @@
from collections.abc import MutableMapping, MutableSequence
from typing import TypeVar
T = TypeVar("T")
type MutableNesting[T] = T | MutableSequence[T] | MutableMapping[T, MutableNesting[T]]
+184 -20
View File
@@ -1,37 +1,73 @@
from dataclasses import dataclass, field from dataclasses import dataclass
from dataclasses import field
from typing import Any
from typing import cast
import pytest
from hooked_containers import events as e from hooked_containers import events as e
from hooked_containers.mapping import HookedMapping from hooked_containers.mapping import HookedMapping
from hooked_containers.mixins import ContainerMixin
from hooked_containers.mixins import ListMixin
from hooked_containers.mixins import MutableMappingMixin
from hooked_containers.mixins import MutableSetMixin
@pytest.mark.parametrize(
("mixin", "storage", "method", "arguments", "expected"),
[
(ListMixin, [1], "append", (2,), e.Change("add", 1, 2)),
(MutableMappingMixin, {"a": 1}, "update", ({"b": 2},), e.Change("add", "b", 2)),
(MutableSetMixin, {1}, "add", (2,), e.Change("add", None, 2)),
],
)
def test_direct_mixin_composition(mixin, storage, method, arguments, expected):
custom_type = type("CustomContainer", (mixin,), {})
container = custom_type()
changes = []
container._storage = storage.copy()
container.hook = changes.append
assert isinstance(container, ContainerMixin)
assert len(container) == 1
assert list(container) == list(storage)
assert next(iter(storage)) in container
assert repr(container) == f"CustomContainer({storage!r})"
assert changes == []
getattr(container, method)(*arguments)
assert changes == [expected]
assert len(container) == 2
assert len(storage) == 1
@dataclass @dataclass
class HookTracker: class HookTracker:
added: list[e.AddItemEvent] = field(default_factory=list) added: list[object] = field(default_factory=list)
set: list[e.SetItemEvent] = field(default_factory=list) set: list[object] = field(default_factory=list)
removed: list[e.RemoveItemEvent] = field(default_factory=list) removed: list[object] = field(default_factory=list)
def hook(self, event: e.ChangeEvent): def hook(self, event: e.Change):
match event: match event.operation:
case e.AddItemEvent(): case "add":
self.added.append(event.item) self.added.append(event.value)
case e.SetItemEvent(): case "set":
self.set.append(event.item) self.set.append(event.value)
case e.RemoveItemEvent(): case "remove":
self.removed.append(event.item) self.removed.append(event.value)
class TestHookedMapping: class TestHookedMapping:
class TestConstruction: class TestConstruction:
def test_empty(self): def test_empty(self):
m = HookedMapping({}) m = HookedMapping({})
assert m._data == {} assert m._storage == {}
assert dict(m) == {} assert dict(m) == {}
def test_with_existing(self): def test_with_existing(self):
existing = {"a": 1, "b": 2} existing = {"a": 1, "b": 2}
og_id = id(existing)
m = HookedMapping(existing) m = HookedMapping(existing)
assert id(m._data) == og_id assert m._storage is not existing
assert dict(m) == existing assert dict(m) == existing
def test_nesting(self): def test_nesting(self):
@@ -41,20 +77,32 @@ class TestHookedMapping:
assert dict(m["a"]) == {"x": 1} assert dict(m["a"]) == {"x": 1}
assert dict(m["b"]) == {"y": 2} assert dict(m["b"]) == {"y": 2}
def test_recreation_copies_storage_without_duplicate_hooks(self):
original_tracker = HookTracker()
tracker = HookTracker()
original = HookedMapping({"a": 1}, original_tracker.hook)
m = HookedMapping(original, tracker.hook)
assert m._storage is not original._storage
m["a"] = 2
assert original._storage == {"a": 1}
assert tracker.set == [2]
assert original_tracker.set == []
class TestMappingOps: class TestMappingOps:
def test_setitem(self): def test_setitem(self):
tracker = HookTracker() tracker = HookTracker()
m = HookedMapping({}, tracker.hook) m = HookedMapping({}, tracker.hook)
m["a"] = 1 m["a"] = 1
assert m._data == {"a": 1} assert m._storage == {"a": 1}
assert tracker.added == [1] assert tracker.added == [1]
def test_nested_setitem(self): def test_nested_values_are_not_observed(self):
tracker = HookTracker() tracker = HookTracker()
m = HookedMapping({"a": {}}, tracker.hook) m = HookedMapping({"a": {}}, tracker.hook)
m["a"]["x"] = 1 m["a"]["x"] = 1
assert m._data == {"a": {"x": 1}} assert m._storage == {"a": {"x": 1}}
assert tracker.added == [1] assert tracker.added == []
def test_getitem(self): def test_getitem(self):
m = HookedMapping({"a": 1}) m = HookedMapping({"a": 1})
@@ -64,5 +112,121 @@ class TestHookedMapping:
tracker = HookTracker() tracker = HookTracker()
m = HookedMapping({"a": 1}, tracker.hook) m = HookedMapping({"a": 1}, tracker.hook)
del m["a"] del m["a"]
assert not m["a"] with pytest.raises(KeyError):
m["a"]
assert tracker.removed == [1] assert tracker.removed == [1]
def test_inherited_mutation_helpers(self):
tracker = HookTracker()
m = HookedMapping({"a": 1}, tracker.hook)
m.update({"a": 2, "b": 3})
assert tracker.set == [2]
assert tracker.added == [3]
assert m.pop("b") == 3
assert tracker.removed == [3]
assert m._storage == {"a": 2}
def test_assignment_returns_none_and_reports_key(self):
changes = []
m = HookedMapping({"a": 1}, changes.append)
assert m.__setitem__("a", 2) is None
assert m.__setitem__("b", 3) is None
assert changes == [e.Change("set", "a", 2), e.Change("add", "b", 3)]
def test_assign_hooked_mapping(self):
changes = []
m = HookedMapping({}, changes.append)
value = HookedMapping({"x": 1})
assert m.__setitem__("a", value) is None
assert m["a"] is value
assert len(changes) == 1
assert changes[0].operation == "add"
assert changes[0].value is value
def test_shallow_value_identity(self):
changes = []
m = HookedMapping({"a": {"x": 1}}, changes.append)
m["a"]["x"] = 2
del m["a"]["x"]
assert changes == []
value = m["a"]
del m["a"]
assert changes == [e.Change("remove", "a", {})]
assert changes[0].value is value
def test_reads_views_and_missing_keys_do_not_mutate(self):
changes = []
mapping = HookedMapping({"a": 1}, changes.append)
keys = mapping.keys()
assert mapping.get("missing", 9) == 9
assert mapping.pop("missing", 9) == 9
with pytest.raises(KeyError):
mapping["missing"]
with pytest.raises(KeyError):
del mapping["missing"]
assert changes == []
assert list(mapping.items()) == [("a", 1)]
mapping["b"] = 2
assert list(keys) == ["a", "b"]
assert list(mapping.values()) == [1, 2]
assert list(reversed(mapping)) == ["b", "a"]
def test_native_order_copy_and_merge(self):
changes = []
mapping = HookedMapping([("a", 1), ("b", 2)], changes.append, c=3)
assert mapping == {"a": 1, "b": 2, "c": 3}
assert mapping.popitem() == ("c", 3)
assert changes == [e.Change("remove", "c", 3)]
changes.clear()
assert type(mapping.copy()) is dict
assert mapping | {"a": 9} == {"a": 9, "b": 2}
assert {"a": 9} | mapping == {"a": 1, "b": 2}
assert changes == []
original = mapping
mapping |= [("b", 4), ("c", 5)]
assert mapping is original
assert changes == [e.Change("set", "b", 4), e.Change("add", "c", 5)]
def test_update_duck_mapping_and_partial_failure(self):
class KeysOnly:
def keys(self):
return ["a"]
def __getitem__(self, _key):
return 1
changes = []
mapping = HookedMapping(hook=changes.append)
mapping.update(KeysOnly())
with pytest.raises(ValueError):
mapping.update(cast(Any, [("b", 2), ("bad",)]))
assert mapping == {"a": 1, "b": 2}
assert changes == [e.Change("add", "a", 1), e.Change("add", "b", 2)]
def test_clear_and_setdefault(self):
changes = []
mapping = HookedMapping[str | None, int]({None: 1}, changes.append)
assert mapping.setdefault(None, 9) == 1
assert changes == []
assert mapping.setdefault("a", 2) == 2
assert changes == [e.Change("add", "a", 2)]
changes.clear()
mapping.clear()
assert mapping == {}
assert changes == [e.Change("remove", "a", 2), e.Change("remove", None, 1)]
changes.clear()
mapping.clear()
assert changes == []
def test_callback_error_leaves_write_in_place(self):
def fail(_change):
raise RuntimeError("callback failed")
mapping = HookedMapping(hook=fail)
with pytest.raises(RuntimeError, match="callback failed"):
mapping["a"] = 1
assert mapping == {"a": 1}
+222 -10
View File
@@ -1,12 +1,20 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Mapping
from collections.abc import MutableMapping
from collections.abc import MutableSequence
from collections.abc import MutableSet
from typing import Any from typing import Any
import pytest
from hooked_containers import events as e
from hooked_containers.mapping import HookedMapping
from hooked_containers.mixins import ListMixin
from hooked_containers.mixins import MutableMappingMixin
from hooked_containers.mixins import MutableSetMixin
from hooked_containers.sequence import HookedList from hooked_containers.sequence import HookedList
from hooked_containers.set import HookedSet
def get_id(a: Any):
return str(hex(id(a)))
class TestHookedList: class TestHookedList:
@@ -21,13 +29,26 @@ class TestHookedList:
def test_recreation(self): def test_recreation(self):
initial = [1, 2, 3] initial = [1, 2, 3]
initial_id = id(initial)
lst = HookedList(initial) lst = HookedList(initial)
assert id(lst._data) == initial_id assert lst._storage is not initial
lst2 = HookedList(lst) lst2 = HookedList(lst)
assert id(lst2._data) == initial_id assert lst2._storage is not lst._storage
assert lst2 == initial
def test_recreation_does_not_duplicate_hooks(self):
original_changes = []
changes = []
original = HookedList([1], original_changes.append)
lst = HookedList(original, changes.append)
lst.append(2)
assert lst._storage is not original._storage
assert original._storage == [1]
assert original_changes == []
assert len(changes) == 1
assert changes == [e.Change("add", 1, 2)]
class TestSeqOps: class TestSeqOps:
def test_len(self): def test_len(self):
@@ -53,16 +74,16 @@ class TestHookedList:
class TestMutableOps: class TestMutableOps:
def test_setitem(self): def test_setitem(self):
added = [] added = []
lst = HookedList( lst = HookedList[Any](
[1, 2, [4, 5, [6, 7]]], [1, 2, [4, 5, [6, 7]]],
lambda e: added.append(e.item), lambda event: added.append(event.value),
) )
lst[0] = 10 lst[0] = 10
lst.append(20) lst.append(20)
assert added == [10, 20] assert added == [10, 20]
lst[2][-1].append(8) lst[2][-1].append(8)
assert added == [10, 20, 8] assert added == [10, 20]
def test_delitem(self): def test_delitem(self):
lst = HookedList([1, 2, 3]) lst = HookedList([1, 2, 3])
@@ -73,3 +94,194 @@ class TestHookedList:
lst = HookedList([1, 3]) lst = HookedList([1, 3])
lst.insert(1, 2) lst.insert(1, 2)
assert list(lst) == [1, 2, 3] assert list(lst) == [1, 2, 3]
def test_inherited_mutation_helpers(self):
changes = []
lst = HookedList([1], changes.append)
lst.extend([2, 3])
assert lst.pop() == 3
lst.remove(1)
assert list(lst) == [2]
assert changes == [
e.Change("add", 1, 2),
e.Change("add", 2, 3),
e.Change("remove", 2, 3),
e.Change("remove", 0, 1),
]
def test_slice_mutations(self):
changes = []
lst = HookedList([1, 2, 3], changes.append)
lst[1:3] = iter([4, 5])
assert list(lst) == [1, 4, 5]
del lst[::2]
assert list(lst) == [4]
assert changes == [
e.Change("set", slice(1, 3), [4, 5]),
e.Change("remove", slice(None, None, 2), [1, 5]),
]
def test_failed_mutations_do_not_notify(self):
changes = []
lst = HookedList([1, 2, 3], changes.append)
with pytest.raises(ValueError):
lst[::2] = [4]
with pytest.raises(IndexError):
del lst[10]
assert list(lst) == [1, 2, 3]
assert changes == []
def test_nested_values_are_not_observed(self):
changes = []
lst = HookedList([[1]], changes.append)
nested = lst[0]
nested[0] = 2
nested.insert(1, 3)
del nested[0]
assert lst._storage == [[3]]
assert changes == []
def test_native_comparisons_and_non_mutating_results(self):
changes = []
items = HookedList([1, 2], changes.append)
assert items == [1, 2]
native = [1, 2]
assert native == items
assert items != (1, 2)
assert items < [1, 3]
assert items <= HookedList([1, 2])
assert items > [0]
assert items >= [1, 2]
with pytest.raises(TypeError):
assert items < (1, 3)
with pytest.raises(TypeError):
_ = items + (3,)
for result in (items[:], items.copy(), items + [3], [0] + items, items * 2, 2 * items):
assert type(result) is list
result.append(99)
assert items == [1, 2]
assert changes == []
def test_index_protocol_and_normalized_indices(self):
class Index:
def __index__(self):
return -1
changes = []
items = HookedList([1, 2], changes.append)
assert items[Index()] == 2
items[Index()] = 3
items.insert(-100, 0)
items.insert(100, 4)
del items[Index()]
assert items == [0, 1, 3]
assert changes == [
e.Change("set", 1, 3),
e.Change("add", 0, 0),
e.Change("add", 3, 4),
e.Change("remove", 3, 4),
]
def test_self_extension_and_in_place_repetition(self):
changes = []
items = HookedList([1, 2], changes.append)
original = items
items += items
assert items is original
assert items == [1, 2, 1, 2]
assert changes == [e.Change("add", 2, 1), e.Change("add", 3, 2)]
changes.clear()
items *= 2
assert items is original
assert items == [1, 2] * 4
assert changes == [e.Change("set", slice(None), [1, 2] * 4)]
changes.clear()
items *= 1
assert changes == []
items *= 0
assert items == []
assert changes == [e.Change("set", slice(None), [])]
def test_reverse_sort_and_empty_writes(self):
changes = []
items = HookedList([3, 1, 2], changes.append)
items.reverse()
assert items == [2, 1, 3]
assert changes == [e.Change("set", 0, 2), e.Change("set", 2, 3)]
changes.clear()
items.sort(reverse=True)
assert items == [3, 2, 1]
assert changes == [e.Change("set", slice(None), [3, 2, 1])]
changes.clear()
del items[0:0]
items[0:0] = []
items.extend([])
assert changes == []
items.clear()
assert [change.value for change in changes] == [1, 2, 3]
changes.clear()
items.clear()
items.reverse()
items.sort()
assert changes == []
def test_partial_extend_and_sort_failure(self):
def values():
yield 2
raise RuntimeError("iteration failed")
changes = []
items = HookedList([1], changes.append)
with pytest.raises(RuntimeError, match="iteration failed"):
items.extend(values())
assert items == [1, 2]
assert changes == [e.Change("add", 1, 2)]
changes.clear()
def fail(_value):
raise ValueError("key failed")
with pytest.raises(ValueError, match="key failed"):
items.sort(key=fail)
assert items == [1, 2]
assert changes == []
@pytest.mark.parametrize(
("mixin", "factory", "mutate", "expected"),
[
(MutableMappingMixin, dict, lambda container: container.update(a=1), e.Change("add", "a", 1)),
(ListMixin, list, lambda container: container.append(1), e.Change("add", 0, 1)),
(MutableSetMixin, set, lambda container: container.add(1), e.Change("add", None, 1)),
],
)
def test_mixins_work_without_concrete_wrappers(mixin, factory, mutate, expected):
class Custom(mixin):
def __init__(self):
self._storage = factory()
changes = []
container = Custom()
container.hook = changes.append
mutate(container)
assert changes == [expected]
def test_helpers_are_inherited_from_standard_abcs():
assert HookedList.append is MutableSequence.append
assert HookedList.extend is MutableSequence.extend
assert HookedList.reverse is MutableSequence.reverse
assert HookedList.__iadd__ is MutableSequence.__iadd__
assert HookedMapping.get is Mapping.get
assert HookedMapping.keys is Mapping.keys
assert HookedMapping.update is MutableMapping.update
assert HookedSet.pop is MutableSet.pop
assert HookedSet.remove is MutableSet.remove
assert HookedSet.__ior__ is MutableSet.__ior__
assert HookedSet.__iand__ is MutableSet.__iand__
assert HookedSet.__ixor__ is MutableSet.__ixor__
assert HookedSet.__isub__ is MutableSet.__isub__
+184
View File
@@ -0,0 +1,184 @@
from dataclasses import FrozenInstanceError
from typing import Any
from typing import cast
import pytest
from hooked_containers import Change
from hooked_containers import HookedSet
class TestHookedSet:
def test_construction_and_owned_storage(self):
changes = []
original = {1, 2}
members = HookedSet(original, changes.append)
assert members == original
assert members._storage is not original
assert HookedSet([1, 1, 2]) == {1, 2}
assert HookedSet() == set()
recreated = HookedSet(members)
assert recreated._storage is not members._storage
assert changes == []
members.add(3)
assert original == {1, 2}
assert recreated == {1, 2}
def test_add_discard_and_no_ops(self):
changes = []
members = HookedSet[str](hook=changes.append)
members.add("a")
members.add("a")
members.discard("missing")
members.discard("a")
assert changes == [Change("add", None, "a"), Change("remove", None, "a")]
changes.clear()
members.clear()
members.update([])
members.difference_update([])
assert changes == []
def test_remove_pop_and_clear(self):
changes = []
members = HookedSet([1, 2], changes.append)
members.remove(1)
assert members.pop() == 2
assert changes == [Change("remove", None, 1), Change("remove", None, 2)]
changes.clear()
with pytest.raises(KeyError):
members.remove(3)
with pytest.raises(KeyError):
members.pop()
assert changes == []
members.update([3, 4])
changes.clear()
members.clear()
assert members == set()
assert {change.value for change in changes} == {3, 4}
assert all(change.operation == "remove" and change.key is None for change in changes)
def test_removal_reports_stored_value(self):
changes = []
members = HookedSet([1], changes.append)
members.discard(True)
assert changes == [Change("remove", None, 1)]
assert type(changes[0].value) is int
def test_hashability_is_validated_before_no_op_detection(self):
changes = []
members = HookedSet([frozenset({1})], changes.append)
with pytest.raises(TypeError):
members.add(cast(Any, {1}))
assert members == {frozenset({1})}
assert changes == []
members.discard(cast(Any, {1}))
assert members == set()
assert changes == [Change("remove", None, frozenset({1}))]
def test_algebra_and_named_helpers_do_not_notify(self):
changes = []
members = HookedSet([1, 2], changes.append)
assert members == {1, 2}
assert members <= {1, 2, 3}
assert members < {1, 2, 3}
assert members >= {1}
assert members > {1}
assert members != {1}
assert members.isdisjoint([3, 4])
assert members.issubset([1, 2, 3])
assert members.issuperset([1])
results = [
(members | {2, 3}, {1, 2, 3}),
(members & {2, 3}, {2}),
(members - {2, 3}, {1}),
(members ^ {2, 3}, {1, 3}),
({3} | members, {1, 2, 3}),
({2, 3} & members, {2}),
({2, 3} - members, {3}),
({2, 3} ^ members, {1, 3}),
(members.union([3], [4]), {1, 2, 3, 4}),
(members.intersection([2, 3]), {2}),
(members.difference([2], [3]), {1}),
(members.symmetric_difference([2, 3]), {1, 3}),
(members.copy(), {1, 2}),
]
for result, expected in results:
assert result == expected
assert type(result) is set
result.add(99)
assert members == {1, 2}
assert changes == []
def test_abc_operators_accept_iterables(self):
members = HookedSet([1, 2])
assert members | cast(Any, [3]) == {1, 2, 3}
assert members & cast(Any, [2]) == {2}
members |= cast(Any, [3])
assert members == {1, 2, 3}
@pytest.mark.parametrize("operation", ["union", "intersection", "difference", "symmetric_difference"])
def test_in_place_operations_keep_identity_and_notify(self, operation):
changes = []
members = HookedSet([1, 2], changes.append)
original = members
if operation == "union":
members |= {2, 3}
expected = {1, 2, 3}
expected_changes = {("add", 3)}
elif operation == "intersection":
members &= {2, 3}
expected = {2}
expected_changes = {("remove", 1)}
elif operation == "difference":
members -= {2, 3}
expected = {1}
expected_changes = {("remove", 2)}
else:
members ^= {2, 3}
expected = {1, 3}
expected_changes = {("remove", 2), ("add", 3)}
assert members is original
assert members == expected
assert {(change.operation, change.value) for change in changes} == expected_changes
assert all(change.key is None for change in changes)
def test_named_updates_and_self_aliases(self):
members = HookedSet([1, 2])
members.update([2, 3], [4])
assert members == {1, 2, 3, 4}
members.intersection_update([2, 3, 4], [3, 4])
assert members == {3, 4}
members.symmetric_difference_update([4, 5, 5])
assert members == {3, 5}
members.difference_update([3], [9])
assert members == {5}
members.update(members)
members.intersection_update(members)
assert members == {5}
members.symmetric_difference_update(members)
assert members == set()
members.update([1, 2])
members.difference_update(members)
assert members == set()
def test_partial_update_failure_is_observed(self):
def values():
yield 1
raise RuntimeError("iteration failed")
changes = []
members = HookedSet[int](hook=changes.append)
with pytest.raises(RuntimeError, match="iteration failed"):
members.update(values())
assert members == {1}
assert changes == [Change("add", None, 1)]
def test_change_metadata_is_immutable_and_values_are_shallow():
value = []
change = Change("set", "key", value)
with pytest.raises(FrozenInstanceError):
cast(Any, change).key = "other"
value.append(1)
assert change.value is value
assert change.value == [1]
Generated
+1233 -47
View File
File diff suppressed because it is too large Load Diff