Compare commits
1
Commits
cbc7e784dc
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
23d3c113d8 |
@@ -1,49 +1,76 @@
|
||||
# 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
|
||||
|
||||
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 |
|
||||
| ---------------- | ------------------------------------------------------------------------------------------- |
|
||||
| Container | Membership testing through `__contains__`. |
|
||||
| Collection | Iteration and size through `__iter__` and `__len__`, alongside membership. |
|
||||
| Hook support | A shared callback and change-event reporting for mutation methods. |
|
||||
| Sequence | Indexed access through `__getitem__`, with sequence helpers such as `index` and `count`. |
|
||||
| Mutable sequence | Assignment, deletion, and insertion, supporting helpers such as `append`, `extend`, and `pop`. |
|
||||
| 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`. |
|
||||
| Mixin | Backing Storage | Responsibility |
|
||||
| --- | --- | --- |
|
||||
| `ContainerMixin` | Any collection | Containment, iteration, length, and representation. |
|
||||
| `HookMixin` | None | Callback dispatch after successful writes. |
|
||||
| `ListMixin` | `list` | Indexed reads and writes, insertion, and native list operations. |
|
||||
| `MutableMappingMixin` | `dict` | Key-based reads and writes, copying, and merging. |
|
||||
| `MutableSetMixin` | `set` | Addition, discard, and named set operations. |
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
from hooked_containers.mapping import HookedMapping
|
||||
from hooked_containers.sequence import HookedList
|
||||
from hooked_containers import HookedList, HookedMapping, HookedSet
|
||||
|
||||
|
||||
def on_change(event):
|
||||
print(type(event).__name__, event.path, event.item)
|
||||
def on_change(change):
|
||||
print(change.operation, change.key, change.value)
|
||||
|
||||
|
||||
items = HookedList(["first"], hook=on_change)
|
||||
items.append("second") # AddItemEvent (1,) second
|
||||
items[0] = "updated" # SetItemEvent (0,) updated
|
||||
del items[1] # RemoveItemEvent (1,) second
|
||||
items.append("second") # add 1 second
|
||||
items[0] = "updated" # set 0 updated
|
||||
del items[1] # remove 1 second
|
||||
|
||||
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.
|
||||
|
||||
@@ -16,6 +16,8 @@ build-backend = "uv_build"
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"ipykernel>=7.2.0",
|
||||
"ipywidgets>=8.1.9",
|
||||
"notebook>=7.6.3",
|
||||
"rich>=14.3.3",
|
||||
"ruff>=0.15.2",
|
||||
"ty>=0.0.85",
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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
|
||||
@@ -1,33 +1,19 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Collection
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Hashable
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
from typing import Protocol
|
||||
from typing import Any
|
||||
from typing import Literal
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ChangeEvent[T]:
|
||||
"""Describe a mutation at an index or key path within a root collection."""
|
||||
class Change[T]:
|
||||
"""Describe a shallow write; removals carry the removed value."""
|
||||
|
||||
root: Collection[object] = field(repr=False)
|
||||
path: Sequence[Hashable | slice]
|
||||
item: T
|
||||
operation: Literal["add", "set", "remove"]
|
||||
key: Hashable | slice | None
|
||||
value: T
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
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: ...
|
||||
type HookFunction = Callable[[Change[Any]], None]
|
||||
|
||||
@@ -1,65 +1,19 @@
|
||||
from collections.abc import MutableMapping
|
||||
from collections.abc import MutableSequence
|
||||
from collections.abc import Iterable
|
||||
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 .mixins import MutableMappingMixin
|
||||
|
||||
|
||||
class HookedMapping[T](MutableMappingMixin[T, MutableNesting[T]], HookedContainer[T]):
|
||||
_data: MutableMapping[T, MutableNesting[T]]
|
||||
class HookedMapping[K, V](MutableMappingMixin[K, V]):
|
||||
"""A shallow mapping with optional mutation notifications."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
existing: MutableMapping[T, MutableNesting[T]],
|
||||
existing: Mapping[K, V] | Iterable[tuple[K, V]] = (),
|
||||
hook: HookFunction | None = None,
|
||||
root: MutableMapping[T] | None = None,
|
||||
path: MutableSequence[int] | None = None,
|
||||
):
|
||||
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)
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self._storage = dict(existing, **kwargs)
|
||||
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
@@ -1,89 +1,65 @@
|
||||
from collections.abc import Collection
|
||||
from collections.abc import Container
|
||||
from collections.abc import Iterable
|
||||
from collections.abc import Iterator
|
||||
from collections.abc import Mapping
|
||||
from collections.abc import MutableMapping
|
||||
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 Literal
|
||||
from typing import Self
|
||||
from typing import SupportsIndex
|
||||
from typing import cast
|
||||
from typing import overload
|
||||
|
||||
from .events import AddItemEvent
|
||||
from .events import ChangeEvent
|
||||
from .events import Change
|
||||
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:
|
||||
"""Share optional mutation callbacks and root-relative event paths."""
|
||||
"""Dispatch shallow changes after successful writes."""
|
||||
|
||||
hook: HookFunction | None = None
|
||||
_root: Any = None
|
||||
_path: Sequence[Any] = ()
|
||||
|
||||
def new_path(self, key: Any) -> list[Any]:
|
||||
return [*self._path, key]
|
||||
|
||||
def _notify(self, event_type: type[ChangeEvent[Any]], key: Any, item: Any) -> None:
|
||||
def _notify(self, operation: Literal["add", "set", "remove"], key: Any, value: Any) -> None:
|
||||
if self.hook is not None:
|
||||
root = self._root if self._root is not None else self
|
||||
self.hook(event_type(root, self.new_path(key), item))
|
||||
self.hook(Change(operation, key, value))
|
||||
|
||||
|
||||
class SequenceMixin[T](CollectionMixin[T], Sequence[T]):
|
||||
"""Add indexed access and the standard read-only sequence helpers."""
|
||||
class ContainerMixin[T](Collection[T]):
|
||||
"""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
|
||||
def __getitem__(self, index: int) -> T: ...
|
||||
def __getitem__(self, index: SupportsIndex) -> T: ...
|
||||
|
||||
@overload
|
||||
def __getitem__(self, index: slice) -> Sequence[T]: ...
|
||||
def __getitem__(self, index: slice) -> list[T]: ...
|
||||
|
||||
def __getitem__(self, index: int | slice) -> T | Sequence[T]:
|
||||
return self._data[index]
|
||||
|
||||
|
||||
class MutableSequenceMixin[T](HookMixin, SequenceMixin[T], MutableSequence[T]):
|
||||
"""Report primitive mutations used by the mutable sequence helpers."""
|
||||
|
||||
_data: MutableSequence[T]
|
||||
def __getitem__(self, index: SupportsIndex | slice) -> T | list[T]:
|
||||
return self._storage[index if isinstance(index, slice) else to_index(index)]
|
||||
|
||||
@overload
|
||||
def __getitem__(self, index: int) -> T: ...
|
||||
|
||||
@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: ...
|
||||
def __setitem__(self, index: SupportsIndex, value: T) -> None: ...
|
||||
|
||||
@overload
|
||||
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:
|
||||
if isinstance(index, slice):
|
||||
value = list(value)
|
||||
self._data[index] = value
|
||||
self._notify(SetItemEvent, index, value)
|
||||
previous = self._storage[index]
|
||||
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:
|
||||
item = self._data[index]
|
||||
del self._data[index]
|
||||
self._notify(RemoveItemEvent, index, item)
|
||||
def __delitem__(self, index: SupportsIndex | slice) -> None:
|
||||
if not isinstance(index, slice):
|
||||
index = to_index(index)
|
||||
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:
|
||||
self._data.insert(index, value)
|
||||
self._notify(AddItemEvent, index, value)
|
||||
def insert(self, index: SupportsIndex, value: T) -> None:
|
||||
index = to_index(index)
|
||||
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]):
|
||||
"""Add key-based access and the standard read-only mapping helpers."""
|
||||
class MutableMappingMixin[K, V](HookMixin, ContainerMixin[K], MutableMapping[K, V]):
|
||||
"""Provide mapping operations and report writes to subclass-owned storage."""
|
||||
|
||||
_data: Mapping[K, V]
|
||||
_storage: dict[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]):
|
||||
"""Report primitive mutations used by the mutable mapping helpers."""
|
||||
def copy(self) -> dict[K, V]:
|
||||
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:
|
||||
event_type = SetItemEvent if key in self._data else AddItemEvent
|
||||
self._data[key] = value
|
||||
self._notify(event_type, key, value)
|
||||
operation: Literal["add", "set"] = "set" if key in self._storage else "add"
|
||||
self._storage[key] = value
|
||||
self._notify(operation, key, value)
|
||||
|
||||
def __delitem__(self, key: K) -> None:
|
||||
item = self._data[key]
|
||||
del self._data[key]
|
||||
self._notify(RemoveItemEvent, key, item)
|
||||
item = self._storage[key]
|
||||
del self._storage[key]
|
||||
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))
|
||||
|
||||
@@ -1,39 +1,12 @@
|
||||
from collections.abc import MutableSequence
|
||||
from collections.abc import Iterable
|
||||
|
||||
from .container import HookedContainer
|
||||
from .events import HookFunction
|
||||
from .mixins import MutableSequenceMixin
|
||||
from .types import MutableNesting
|
||||
from .mixins import ListMixin
|
||||
|
||||
|
||||
class HookedList[T](MutableSequenceMixin[T], HookedContainer[T]):
|
||||
_data: MutableSequence[T]
|
||||
path: MutableSequence[int]
|
||||
class HookedList[T](ListMixin[T]):
|
||||
"""A shallow list with optional mutation notifications."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
existing: MutableSequence[T],
|
||||
hook: HookFunction | None = None,
|
||||
root: MutableNesting[T] | None = None,
|
||||
path: MutableSequence[int] | None = None,
|
||||
) -> None:
|
||||
def __init__(self, existing: Iterable[T] = (), hook: HookFunction | None = None) -> None:
|
||||
self._storage = list(existing)
|
||||
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
|
||||
|
||||
@@ -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
|
||||
@@ -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)}")
|
||||
@@ -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
@@ -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.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
|
||||
class HookTracker:
|
||||
added: list[e.AddItemEvent] = field(default_factory=list)
|
||||
set: list[e.SetItemEvent] = field(default_factory=list)
|
||||
removed: list[e.RemoveItemEvent] = field(default_factory=list)
|
||||
added: list[object] = field(default_factory=list)
|
||||
set: list[object] = field(default_factory=list)
|
||||
removed: list[object] = field(default_factory=list)
|
||||
|
||||
def hook(self, event: e.ChangeEvent):
|
||||
match event:
|
||||
case e.AddItemEvent():
|
||||
self.added.append(event.item)
|
||||
case e.SetItemEvent():
|
||||
self.set.append(event.item)
|
||||
case e.RemoveItemEvent():
|
||||
self.removed.append(event.item)
|
||||
def hook(self, event: e.Change):
|
||||
match event.operation:
|
||||
case "add":
|
||||
self.added.append(event.value)
|
||||
case "set":
|
||||
self.set.append(event.value)
|
||||
case "remove":
|
||||
self.removed.append(event.value)
|
||||
|
||||
|
||||
class TestHookedMapping:
|
||||
class TestConstruction:
|
||||
def test_empty(self):
|
||||
m = HookedMapping({})
|
||||
assert m._data == {}
|
||||
assert m._storage == {}
|
||||
assert dict(m) == {}
|
||||
|
||||
def test_with_existing(self):
|
||||
existing = {"a": 1, "b": 2}
|
||||
og_id = id(existing)
|
||||
m = HookedMapping(existing)
|
||||
assert id(m._data) == og_id
|
||||
assert m._storage is not existing
|
||||
assert dict(m) == existing
|
||||
|
||||
def test_nesting(self):
|
||||
@@ -41,20 +77,32 @@ class TestHookedMapping:
|
||||
assert dict(m["a"]) == {"x": 1}
|
||||
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:
|
||||
def test_setitem(self):
|
||||
tracker = HookTracker()
|
||||
m = HookedMapping({}, tracker.hook)
|
||||
m["a"] = 1
|
||||
assert m._data == {"a": 1}
|
||||
assert m._storage == {"a": 1}
|
||||
assert tracker.added == [1]
|
||||
|
||||
def test_nested_setitem(self):
|
||||
def test_nested_values_are_not_observed(self):
|
||||
tracker = HookTracker()
|
||||
m = HookedMapping({"a": {}}, tracker.hook)
|
||||
m["a"]["x"] = 1
|
||||
assert m._data == {"a": {"x": 1}}
|
||||
assert tracker.added == [1]
|
||||
assert m._storage == {"a": {"x": 1}}
|
||||
assert tracker.added == []
|
||||
|
||||
def test_getitem(self):
|
||||
m = HookedMapping({"a": 1})
|
||||
@@ -64,5 +112,121 @@ class TestHookedMapping:
|
||||
tracker = HookTracker()
|
||||
m = HookedMapping({"a": 1}, tracker.hook)
|
||||
del m["a"]
|
||||
assert not m["a"]
|
||||
with pytest.raises(KeyError):
|
||||
m["a"]
|
||||
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
@@ -1,12 +1,20 @@
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
|
||||
def get_id(a: Any):
|
||||
return str(hex(id(a)))
|
||||
from hooked_containers.set import HookedSet
|
||||
|
||||
|
||||
class TestHookedList:
|
||||
@@ -21,13 +29,26 @@ class TestHookedList:
|
||||
|
||||
def test_recreation(self):
|
||||
initial = [1, 2, 3]
|
||||
initial_id = id(initial)
|
||||
|
||||
lst = HookedList(initial)
|
||||
assert id(lst._data) == initial_id
|
||||
assert lst._storage is not initial
|
||||
|
||||
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:
|
||||
def test_len(self):
|
||||
@@ -53,16 +74,16 @@ class TestHookedList:
|
||||
class TestMutableOps:
|
||||
def test_setitem(self):
|
||||
added = []
|
||||
lst = HookedList(
|
||||
lst = HookedList[Any](
|
||||
[1, 2, [4, 5, [6, 7]]],
|
||||
lambda e: added.append(e.item),
|
||||
lambda event: added.append(event.value),
|
||||
)
|
||||
lst[0] = 10
|
||||
lst.append(20)
|
||||
assert added == [10, 20]
|
||||
|
||||
lst[2][-1].append(8)
|
||||
assert added == [10, 20, 8]
|
||||
assert added == [10, 20]
|
||||
|
||||
def test_delitem(self):
|
||||
lst = HookedList([1, 2, 3])
|
||||
@@ -73,3 +94,194 @@ class TestHookedList:
|
||||
lst = HookedList([1, 3])
|
||||
lst.insert(1, 2)
|
||||
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__
|
||||
|
||||
@@ -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]
|
||||
Reference in New Issue
Block a user