From c10f6286ee85f066f313f2d14a2fb43a54c1679b Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 00:03:46 -0500 Subject: [PATCH 01/11] add snapshot lifecycle hooks --- src/snapclass/collections.py | 17 +++- src/snapclass/schemas.py | 76 ++++++++++++++++++ src/snapclass/snapshots.py | 3 +- tests/test_behavior_contracts.py | 134 ++++++++++++++++++++++++++++++- tests/test_sidecar.py | 18 +++++ 5 files changed, 242 insertions(+), 6 deletions(-) diff --git a/src/snapclass/collections.py b/src/snapclass/collections.py index 8ce74de..aa34b38 100644 --- a/src/snapclass/collections.py +++ b/src/snapclass/collections.py @@ -27,11 +27,12 @@ def __call__(self, stash: Stash | str | os.PathLike[str]) -> "Collection": return Collection(self.model, _coerce_stash(stash)) def get(self, *args: Any, **kwargs: Any) -> Any: - from .schemas import _attach_snapshot + from .schemas import _attach_snapshot, _mark_snapshot_ready __tracebackhide__ = sessions.HIDDEN_TRACEBACK instance = self._empty_instance(*args, **kwargs) _attach_snapshot(instance, self.model.__snapclass_config__, self._stash) + _mark_snapshot_ready(instance) instance.snapshot.load(_initial=True) return instance @@ -43,11 +44,12 @@ def get_or_none(self, *args: Any, **kwargs: Any) -> Any | None: return None def get_or_create(self, *args: Any, **kwargs: Any) -> Any: - from .schemas import _attach_snapshot, _write_lock_for + from .schemas import _attach_snapshot, _mark_snapshot_ready, _write_lock_for __tracebackhide__ = sessions.HIDDEN_TRACEBACK instance = self._empty_instance(*args, **kwargs, include_defaults=True) _attach_snapshot(instance, self.model.__snapclass_config__, self._stash) + _mark_snapshot_ready(instance) with _write_lock_for(instance.snapshot._require_path()): if instance.snapshot.exists: instance.snapshot.load(_initial=True) @@ -56,7 +58,12 @@ def get_or_create(self, *args: Any, **kwargs: Any) -> Any: return instance def all(self, *, _exclude: str = "") -> Iterator[Any]: - from .schemas import _PatternMatcher, _attach_snapshot, _has_path_value + from .schemas import ( + _PatternMatcher, + _attach_snapshot, + _has_path_value, + _mark_snapshot_ready, + ) __tracebackhide__ = sessions.HIDDEN_TRACEBACK if not self.model.__snapclass_config__.pattern: @@ -76,7 +83,9 @@ def all(self, *, _exclude: str = "") -> Iterator[Any]: if matcher.has_recursive_wildcard or _has_path_value(values): instance = self._empty_instance(*values) _attach_snapshot(instance, self.model.__snapclass_config__, self._stash) - instance.snapshot.load(path, _initial=True) + instance.snapshot.path = path + _mark_snapshot_ready(instance) + instance.snapshot.load(_initial=True) yield instance else: yield self.get(*values) diff --git a/src/snapclass/schemas.py b/src/snapclass/schemas.py index 94ccbb1..586f70e 100644 --- a/src/snapclass/schemas.py +++ b/src/snapclass/schemas.py @@ -355,6 +355,7 @@ def sync( if not hasattr(cls, "__snapclass_config__"): _install(cls, config) _attach_snapshot(instance, config) + _mark_snapshot_ready(instance) if _auto_enabled(config): instance.snapshot.save() return instance @@ -428,6 +429,79 @@ def frozen(*snapshots: object): snapshot.save() +def _mark_snapshot_ready(instance: object) -> None: + """Run ``__snapclass_ready__`` after snapshot attachment and setup settle.""" + snapshot = getattr(instance, "snapshot", None) + if snapshot is None: + return + _call_snapshot_lifecycle_hook( + instance, + "__snapclass_ready__", + snapshot=snapshot, + path=_snapshot_path_or_none(snapshot), + ) + + +def _mark_snapshot_loaded(instance: object, path: Path) -> None: + """Run ``__snapclass_loaded__`` after file data is applied and tracked.""" + snapshot = getattr(instance, "snapshot", None) + if snapshot is None: + return + _call_snapshot_lifecycle_hook( + instance, + "__snapclass_loaded__", + snapshot=snapshot, + path=path, + ) + + +def _call_snapshot_lifecycle_hook( + instance: object, + hook_name: str, + *, + snapshot: "Snapshot", + path: Path | None = None, +) -> None: + """Invoke a snapclass lifecycle hook with suppressed autosave/reload hooks.""" + hook = getattr(instance, hook_name, None) + if hook is None: + return + try: + with _snapclass_hook_context(instance): + if hook_name == "__snapclass_loaded__": + hook(snapshot=snapshot, path=path) + else: + hook(snapshot=snapshot) + except Exception as exc: + location = f" at {path}" if path is not None else "" + raise SnapclassError( + f"Failed to run {hook_name} for {instance.__class__.__name__}" + f"{location}: {exc}" + ) from exc + + +@contextmanager +def _snapclass_hook_context(instance: object) -> Iterator[None]: + """Suppress automatic snapclass saves and reloads during lifecycle hooks.""" + previous_hooks = sessions.HOOKS_ENABLED + previous_loading = getattr(instance, "_snapclass_loading", False) + sessions.HOOKS_ENABLED = False + object.__setattr__(instance, "_snapclass_loading", True) + try: + yield + finally: + object.__setattr__(instance, "_snapclass_loading", previous_loading) + sessions.HOOKS_ENABLED = previous_hooks + + +def _snapshot_path_or_none(snapshot: "Snapshot") -> Path | None: + """Return the resolved snapshot path when it can be used for diagnostics.""" + try: + return snapshot.path + except Exception: + return None + + def _install(cls: type, config: Config) -> None: cls.__snapclass_config__ = config cls.snapshots = CollectionDescriptor() @@ -449,6 +523,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: _apply_sidecar_values(self, pending_sidecars, save_metadata=False) _apply_sidecar_values(self, sidecar_values, save_metadata=False) object.__setattr__(self, "_snapclass_initializing", False) + _mark_snapshot_ready(self) if _auto_enabled(config): if self.snapshot.exists: self.snapshot.load(_initial=True) @@ -794,6 +869,7 @@ def load( self._loaded_path = current_path self._last_text = text self.modified = False + _mark_snapshot_loaded(self._instance, current_path) finally: object.__setattr__(self._instance, "_snapclass_loading", False) diff --git a/src/snapclass/snapshots.py b/src/snapclass/snapshots.py index 8792e09..86ddd08 100644 --- a/src/snapclass/snapshots.py +++ b/src/snapclass/snapshots.py @@ -2,7 +2,7 @@ from typing import Any -from .schemas import Snapshot, _attach_snapshot +from .schemas import Snapshot, _attach_snapshot, _mark_snapshot_ready def create_snapshot(obj: Any, root: Snapshot | None = None) -> Snapshot: @@ -11,6 +11,7 @@ def create_snapshot(obj: Any, root: Snapshot | None = None) -> Snapshot: return snapshot config = obj.__class__.__snapclass_config__ _attach_snapshot(obj, config, root.stash if root is not None else None) + _mark_snapshot_ready(obj) return obj.snapshot diff --git a/tests/test_behavior_contracts.py b/tests/test_behavior_contracts.py index bfedca1..1e7267e 100644 --- a/tests/test_behavior_contracts.py +++ b/tests/test_behavior_contracts.py @@ -6,7 +6,7 @@ import pytest from ruamel.yaml import YAML as RuamelYAML -from snapclass import Stash, snapclass, hooks +from snapclass import SnapclassError, Stash, snapclass, hooks def test_snapclass_types_behave_like_plain_yaml_values(): @@ -150,3 +150,135 @@ class Item: "events:\n" " - nested\n" ) + + +def test_snapshots_get_initializes_init_false_fields_before_loaded(tmp_path): + @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) + class Item: + name: str + value: str = "" + transient: list[str] = field(init=False) + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and transient state can be initialized.""" + self.transient = ["ready"] + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied and transient state can be reused.""" + self.transient.append(f"loaded:{path.name}") + + (tmp_path / "sample.yml").write_text("value: file\n", encoding="utf-8") + + item = Item.snapshots.get("sample") + + assert item.value == "file" + assert item.transient == ["ready", "loaded:sample.yml"] + + +def test_snapshots_get_or_create_missing_file_runs_ready_without_loaded(tmp_path): + @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) + class Item: + name: str + value: str = "created" + transient: list[str] = field(init=False) + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and transient state can be initialized.""" + self.transient = ["ready"] + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied and transient state can be reused.""" + self.transient.append("loaded") + + item = Item.snapshots.get_or_create("sample") + + assert item.value == "created" + assert item.transient == ["ready"] + assert (tmp_path / "sample.yml").exists() + + +def test_loaded_hook_runs_for_explicit_loads_and_text_setter(tmp_path): + @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) + class Item: + name: str + value: str = "" + loaded_paths: list[str] = field(default_factory=list) + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied and transient state can be reused.""" + self.loaded_paths.append(path.name) + + first = Item("first") + (tmp_path / "first.yml").write_text("value: snapshot\n", encoding="utf-8") + first.snapshot.load() + + second = Item("second") + (tmp_path / "second.yml").write_text("value: object\n", encoding="utf-8") + second.load() + + third = Item("third") + third.snapshot.save() + third.snapshot.text = "value: text\n" + + assert first.loaded_paths == ["first.yml"] + assert second.loaded_paths == ["second.yml"] + assert third.loaded_paths == ["third.yml"] + assert first.value == "snapshot" + assert second.value == "object" + assert third.value == "text" + + +def test_lifecycle_hook_mutations_do_not_trigger_automatic_save_loops(tmp_path): + @snapclass("{self.name}.yml", stash=Stash(tmp_path)) + class Item: + name: str + value: str = "" + events: list[str] = field(default_factory=list) + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied and transient state can be reused.""" + self.value = "hooked" + self.events.append("loaded") + + path = tmp_path / "sample.yml" + path.write_text("value: file\nevents: []\n", encoding="utf-8") + + item = Item("sample") + + assert item.value == "hooked" + assert item.events == ["loaded"] + assert path.read_text(encoding="utf-8") == "value: file\nevents: []\n" + + +def test_lifecycle_hook_failures_include_hook_name_and_path(tmp_path): + @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) + class ReadyItem: + name: str + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and transient state can be initialized.""" + raise RuntimeError("ready failed") + + with pytest.raises(SnapclassError) as ready_error: + ReadyItem("ready") + + ready_message = str(ready_error.value) + assert "__snapclass_ready__" in ready_message + assert "ready.yml" in ready_message + + @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) + class LoadedItem: + name: str + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied and transient state can be reused.""" + raise RuntimeError("loaded failed") + + (tmp_path / "loaded.yml").write_text("", encoding="utf-8") + + with pytest.raises(SnapclassError) as loaded_error: + LoadedItem.snapshots.get("loaded") + + loaded_message = str(loaded_error.value) + assert "__snapclass_loaded__" in loaded_message + assert "loaded.yml" in loaded_message diff --git a/tests/test_sidecar.py b/tests/test_sidecar.py index 3f099c0..c109f63 100644 --- a/tests/test_sidecar.py +++ b/tests/test_sidecar.py @@ -38,6 +38,24 @@ class Article: assert article.body.snapshot.stash == articles +def test_sidecar_constructor_values_are_visible_before_ready_hook_runs(tmp_path): + articles = Stash(tmp_path / "world") / "article" + observed: list[str] = [] + + @snapclass("{self.slug}/article.yml", stash=articles, manual=True) + class Article: + slug: str + body: str = sidecar.text("{self.slug}.md") + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and sidecar constructor values are visible.""" + observed.append(self.body) + + Article("dusk-court", body="# Dusk Court\n") + + assert observed == ["# Dusk Court\n"] + + def test_text_sidecar_can_use_explicit_stash(tmp_path): app = Stash(tmp_path / "world") articles = app / "article" From 4fc4688f1dc8e472667d78667e604504e42d426f Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 00:27:07 -0500 Subject: [PATCH 02/11] address lifecycle hook review feedback --- src/snapclass/collections.py | 7 ++-- src/snapclass/hooks.py | 2 ++ src/snapclass/schemas.py | 45 +++++++++++++------------ tests/test_behavior_contracts.py | 56 +++++++++++++++++++++++++++++++- 4 files changed, 84 insertions(+), 26 deletions(-) diff --git a/src/snapclass/collections.py b/src/snapclass/collections.py index aa34b38..f6d3e6f 100644 --- a/src/snapclass/collections.py +++ b/src/snapclass/collections.py @@ -27,12 +27,11 @@ def __call__(self, stash: Stash | str | os.PathLike[str]) -> "Collection": return Collection(self.model, _coerce_stash(stash)) def get(self, *args: Any, **kwargs: Any) -> Any: - from .schemas import _attach_snapshot, _mark_snapshot_ready + from .schemas import _attach_snapshot __tracebackhide__ = sessions.HIDDEN_TRACEBACK instance = self._empty_instance(*args, **kwargs) _attach_snapshot(instance, self.model.__snapclass_config__, self._stash) - _mark_snapshot_ready(instance) instance.snapshot.load(_initial=True) return instance @@ -49,11 +48,11 @@ def get_or_create(self, *args: Any, **kwargs: Any) -> Any: __tracebackhide__ = sessions.HIDDEN_TRACEBACK instance = self._empty_instance(*args, **kwargs, include_defaults=True) _attach_snapshot(instance, self.model.__snapclass_config__, self._stash) - _mark_snapshot_ready(instance) with _write_lock_for(instance.snapshot._require_path()): if instance.snapshot.exists: instance.snapshot.load(_initial=True) else: + _mark_snapshot_ready(instance) instance.snapshot.save() return instance @@ -62,7 +61,6 @@ def all(self, *, _exclude: str = "") -> Iterator[Any]: _PatternMatcher, _attach_snapshot, _has_path_value, - _mark_snapshot_ready, ) __tracebackhide__ = sessions.HIDDEN_TRACEBACK @@ -84,7 +82,6 @@ def all(self, *, _exclude: str = "") -> Iterator[Any]: instance = self._empty_instance(*values) _attach_snapshot(instance, self.model.__snapclass_config__, self._stash) instance.snapshot.path = path - _mark_snapshot_ready(instance) instance.snapshot.load(_initial=True) yield instance else: diff --git a/src/snapclass/hooks.py b/src/snapclass/hooks.py index 33333fa..f93cb7f 100644 --- a/src/snapclass/hooks.py +++ b/src/snapclass/hooks.py @@ -38,6 +38,8 @@ def enabled(snapshot: Any, args: Iterable[Any]) -> bool: return False if getattr(snapshot, "manual", False): return False + if getattr(getattr(snapshot, "_instance", None), "_snapclass_hooks_suppressed", False): + return False name = _first_string_arg(args) if name is None: return True diff --git a/src/snapclass/schemas.py b/src/snapclass/schemas.py index 586f70e..7f1df2f 100644 --- a/src/snapclass/schemas.py +++ b/src/snapclass/schemas.py @@ -356,7 +356,7 @@ def sync( _install(cls, config) _attach_snapshot(instance, config) _mark_snapshot_ready(instance) - if _auto_enabled(config): + if _auto_enabled(config, instance): instance.snapshot.save() return instance @@ -482,16 +482,16 @@ def _call_snapshot_lifecycle_hook( @contextmanager def _snapclass_hook_context(instance: object) -> Iterator[None]: - """Suppress automatic snapclass saves and reloads during lifecycle hooks.""" - previous_hooks = sessions.HOOKS_ENABLED + """Suppress automatic saves and reloads for one instance during lifecycle hooks.""" previous_loading = getattr(instance, "_snapclass_loading", False) - sessions.HOOKS_ENABLED = False + previous_suppressed = getattr(instance, "_snapclass_hooks_suppressed", False) object.__setattr__(instance, "_snapclass_loading", True) + object.__setattr__(instance, "_snapclass_hooks_suppressed", True) try: yield finally: + object.__setattr__(instance, "_snapclass_hooks_suppressed", previous_suppressed) object.__setattr__(instance, "_snapclass_loading", previous_loading) - sessions.HOOKS_ENABLED = previous_hooks def _snapshot_path_or_none(snapshot: "Snapshot") -> Path | None: @@ -523,11 +523,11 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: _apply_sidecar_values(self, pending_sidecars, save_metadata=False) _apply_sidecar_values(self, sidecar_values, save_metadata=False) object.__setattr__(self, "_snapclass_initializing", False) - _mark_snapshot_ready(self) - if _auto_enabled(config): - if self.snapshot.exists: - self.snapshot.load(_initial=True) - else: + if _auto_enabled(config, self) and self.snapshot.exists: + self.snapshot.load(_initial=True) + else: + _mark_snapshot_ready(self) + if _auto_enabled(config, self): self.snapshot.save() cls.__init__ = __init__ @@ -555,7 +555,7 @@ def __setattr__(self, name: str, value: Any) -> None: ) if ( should_track - and _auto_enabled(snapshot._config) + and _auto_enabled(snapshot._config, self) and not getattr(self, "_snapclass_loading", False) and snapshot.exists and snapshot.modified @@ -588,7 +588,7 @@ def __setattr__(self, name: str, value: Any) -> None: if inferred_hint is not None: inferred_hints[name] = inferred_hint object.__setattr__(self, _INFERRED_HINTS_ATTR, inferred_hints) - if _auto_enabled(snapshot._config): + if _auto_enabled(snapshot._config, self): snapshot.save() snapshot.load() @@ -606,7 +606,7 @@ def __getattribute__(self, name: str) -> Any: if ( name in field_names and snapshot is not None - and _auto_enabled(config) + and _auto_enabled(config, self) and not object.__getattribute__(self, "__dict__").get("_snapclass_initializing", False) and not object.__getattribute__(self, "__dict__").get("_snapclass_loading", False) and snapshot.exists @@ -869,6 +869,7 @@ def load( self._loaded_path = current_path self._last_text = text self.modified = False + _mark_snapshot_ready(self._instance) _mark_snapshot_loaded(self._instance, current_path) finally: object.__setattr__(self._instance, "_snapclass_loading", False) @@ -910,7 +911,7 @@ def __init__(self, values: list[Any], snapshot: Snapshot) -> None: super().__init__(_track_value(value, snapshot) for value in values) def _save(self) -> None: - if _auto_enabled(self._snapshot._config): + if _auto_enabled(self._snapshot._config, self._snapshot._instance): self._snapshot.save() _coerce_tracked_container_in_place(self, self._snapshot) @@ -971,7 +972,7 @@ def __init__(self, values: Any, snapshot: Snapshot) -> None: super().__init__(_track_value(value, snapshot) for value in values) def _save(self) -> None: - if _auto_enabled(self._snapshot._config): + if _auto_enabled(self._snapshot._config, self._snapshot._instance): self._snapshot.save() _coerce_tracked_container_in_place(self, self._snapshot) @@ -1041,7 +1042,7 @@ def __init__(self, values: Any, snapshot: Snapshot) -> None: Counter.update(self, values) def _save(self) -> None: - if _auto_enabled(self._snapshot._config): + if _auto_enabled(self._snapshot._config, self._snapshot._instance): self._snapshot.save() _coerce_tracked_container_in_place(self, self._snapshot) @@ -1106,7 +1107,7 @@ def __delattr__(self, name: str) -> None: raise AttributeError(name) from exc def _save(self) -> None: - if _auto_enabled(self._snapshot._config): + if _auto_enabled(self._snapshot._config, self._snapshot._instance): self._snapshot.save() _coerce_tracked_container_in_place(self, self._snapshot) @@ -1192,7 +1193,7 @@ def __delattr__(self, name: str) -> None: raise AttributeError(name) from exc def _save(self) -> None: - if _auto_enabled(self._snapshot._config): + if _auto_enabled(self._snapshot._config, self._snapshot._instance): self._snapshot.save() _coerce_tracked_container_in_place(self, self._snapshot) @@ -2660,8 +2661,12 @@ def _replace_path_atomic(temp_path: Path, path: Path) -> None: time.sleep(0.01 * (attempt + 1)) -def _auto_enabled(config: Config) -> bool: - return sessions.HOOKS_ENABLED and not config.manual +def _auto_enabled(config: Config, instance: object | None = None) -> bool: + if not sessions.HOOKS_ENABLED or config.manual: + return False + if instance is not None and getattr(instance, "_snapclass_hooks_suppressed", False): + return False + return True def _lookup(item: object, key: str) -> Any: diff --git a/tests/test_behavior_contracts.py b/tests/test_behavior_contracts.py index 1e7267e..4ca7934 100644 --- a/tests/test_behavior_contracts.py +++ b/tests/test_behavior_contracts.py @@ -6,7 +6,7 @@ import pytest from ruamel.yaml import YAML as RuamelYAML -from snapclass import SnapclassError, Stash, snapclass, hooks +from snapclass import SnapclassError, Stash, snapclass, hooks, sessions def test_snapclass_types_behave_like_plain_yaml_values(): @@ -250,6 +250,60 @@ def __snapclass_loaded__(self, *, snapshot, path): assert path.read_text(encoding="utf-8") == "value: file\nevents: []\n" +def test_ready_hook_does_not_mask_file_data_during_initial_load(tmp_path): + @snapclass("{self.name}.yml", stash=Stash(tmp_path)) + class Item: + name: str + value: str = "" + ready_seen: str = field(init=False) + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and file data is visible when present.""" + self.ready_seen = self.value + if not self.value: + self.value = "ready" + + (tmp_path / "sample.yml").write_text("value: file\n", encoding="utf-8") + + item = Item("sample") + + assert item.ready_seen == "file" + assert item.value == "file" + assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" + + +def test_lifecycle_hook_suppression_is_per_instance(tmp_path): + @snapclass("{self.name}.yml", stash=Stash(tmp_path / "other")) + class Other: + name: str + value: str = "" + + other = Other("target") + + @snapclass("{self.name}.yml", stash=Stash(tmp_path / "items")) + class Item: + name: str + value: str = "" + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied and only this instance is suppressed.""" + assert sessions.HOOKS_ENABLED is True + self.value = "hooked" + other.value = "updated" + + item_path = tmp_path / "items" / "sample.yml" + item_path.parent.mkdir() + item_path.write_text("value: file\n", encoding="utf-8") + + item = Item("sample") + + assert item.value == "hooked" + assert item_path.read_text(encoding="utf-8") == "value: file\n" + assert (tmp_path / "other" / "target.yml").read_text(encoding="utf-8") == ( + "value: updated\n" + ) + + def test_lifecycle_hook_failures_include_hook_name_and_path(tmp_path): @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) class ReadyItem: From 67f22d37b0d11b46624e25f8a494c7e93b5184e3 Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 01:04:24 -0500 Subject: [PATCH 03/11] tighten lifecycle hook suppression --- src/snapclass/schemas.py | 25 ++++++++++++---- src/snapclass/sidecar.py | 51 ++++++++++++++++++++++++++++++++ tests/test_behavior_contracts.py | 26 ++++++++++++++++ tests/test_sidecar.py | 34 +++++++++++++++++++++ 4 files changed, 130 insertions(+), 6 deletions(-) diff --git a/src/snapclass/schemas.py b/src/snapclass/schemas.py index 7f1df2f..765cc6a 100644 --- a/src/snapclass/schemas.py +++ b/src/snapclass/schemas.py @@ -434,12 +434,19 @@ def _mark_snapshot_ready(instance: object) -> None: snapshot = getattr(instance, "snapshot", None) if snapshot is None: return - _call_snapshot_lifecycle_hook( - instance, - "__snapclass_ready__", - snapshot=snapshot, - path=_snapshot_path_or_none(snapshot), - ) + if getattr(snapshot, "_ready", False): + return + snapshot._ready = True + try: + _call_snapshot_lifecycle_hook( + instance, + "__snapclass_ready__", + snapshot=snapshot, + path=_snapshot_path_or_none(snapshot), + ) + except Exception: + snapshot._ready = False + raise def _mark_snapshot_loaded(instance: object, path: Path) -> None: @@ -544,6 +551,10 @@ def __setattr__(self, name: str, value: Any) -> None: pending[name] = value object.__setattr__(self, _PENDING_SIDECARS_ATTR, pending) return + if getattr(self, "_snapclass_hooks_suppressed", False): + sidecar_descriptor._set_override(self, value) + return + sidecar_descriptor._clear_override(self) sidecar_descriptor.snapshot(self).write(value) return snapshot = getattr(self, "snapshot", None) @@ -680,6 +691,7 @@ def __init__( self._last_mtime: float | None = None self._loaded_data: dict[str, Any] | None = None self._loaded_path: Path | None = None + self._ready = False @property def classname(self) -> str: @@ -861,6 +873,7 @@ def load( object.__setattr__(self._instance, "_snapclass_loading", True) try: try: + sidecar.clear_overrides(self._instance) _apply_data(self._instance, data, preserve_non_default=_initial) except _CoercionError as exc: raise SnapclassError(_schema_mismatch_message(current_path, exc)) from exc diff --git a/src/snapclass/sidecar.py b/src/snapclass/sidecar.py index b1bc217..32a6308 100644 --- a/src/snapclass/sidecar.py +++ b/src/snapclass/sidecar.py @@ -12,6 +12,7 @@ _Kind = Literal["text", "bytes"] _StashLike = Stash | str | os.PathLike[str] _MISSING = object() +_OVERRIDES_ATTR = "__snapclass_sidecar_overrides__" class SidecarMissingError(FileNotFoundError): @@ -83,6 +84,10 @@ class SidecarDescriptor: encoding: str = "utf-8" stash: Stash | None = None + def __set_name__(self, owner: type, name: str) -> None: + """Remember the model attribute name for suppressed in-memory values.""" + object.__setattr__(self, "_name", name) + def __get__(self, instance: object | None, owner: type | None = None): if instance is None: return self @@ -96,10 +101,56 @@ def snapshot(self, instance: object) -> "SidecarSnapshot": def value(self, instance: object) -> "SidecarText | SidecarBytes": snapshot = self.snapshot(instance) + override = self._override_value(instance) + if override is not _MISSING: + if self.kind == "text": + return SidecarText(cast(str, override), snapshot) + return SidecarBytes(cast(builtins.bytes, override), snapshot) if self.kind == "text": return SidecarText(snapshot.read(default=""), snapshot) return SidecarBytes(snapshot.read(default=b""), snapshot) + def _set_override(self, instance: object, value: str | builtins.bytes) -> None: + """Store a sidecar assignment in memory without writing sidecar files.""" + if self.kind == "text" and not isinstance(value, str): + raise TypeError("Text sidecars require str values") + if self.kind == "bytes" and not isinstance( + value, + (builtins.bytes, bytearray, memoryview), + ): + raise TypeError("Bytes sidecars require bytes-like values") + name = getattr(self, "_name", None) + if name is None: + return + overrides = dict(getattr(instance, _OVERRIDES_ATTR, {})) + overrides[name] = ( + builtins.bytes(value) if self.kind == "bytes" else value + ) + object.__setattr__(instance, _OVERRIDES_ATTR, overrides) + + def _clear_override(self, instance: object) -> None: + """Remove an in-memory sidecar assignment for this descriptor.""" + name = getattr(self, "_name", None) + if name is None: + return + overrides = dict(getattr(instance, _OVERRIDES_ATTR, {})) + if name not in overrides: + return + del overrides[name] + object.__setattr__(instance, _OVERRIDES_ATTR, overrides) + + def _override_value(self, instance: object) -> object: + """Return a suppressed in-memory value or the missing sentinel.""" + name = getattr(self, "_name", None) + if name is None: + return _MISSING + return getattr(instance, _OVERRIDES_ATTR, {}).get(name, _MISSING) + + +def clear_overrides(instance: object) -> None: + """Clear all suppressed sidecar assignments for an instance.""" + object.__setattr__(instance, _OVERRIDES_ATTR, {}) + class SidecarText(str): snapshot: "SidecarSnapshot" diff --git a/tests/test_behavior_contracts.py b/tests/test_behavior_contracts.py index 4ca7934..c0b51b8 100644 --- a/tests/test_behavior_contracts.py +++ b/tests/test_behavior_contracts.py @@ -272,6 +272,32 @@ def __snapclass_ready__(self, *, snapshot): assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" +def test_ready_hook_runs_once_but_loaded_runs_on_each_load(tmp_path): + calls: list[str] = [] + + @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) + class Item: + name: str + value: str = "" + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and ready runs once per attached snapshot.""" + calls.append("ready") + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied and loaded runs for each load.""" + calls.append("loaded") + + (tmp_path / "sample.yml").write_text("value: first\n", encoding="utf-8") + item = Item.snapshots.get("sample") + (tmp_path / "sample.yml").write_text("value: second\n", encoding="utf-8") + + item.snapshot.load() + + assert item.value == "second" + assert calls == ["ready", "loaded", "loaded"] + + def test_lifecycle_hook_suppression_is_per_instance(tmp_path): @snapclass("{self.name}.yml", stash=Stash(tmp_path / "other")) class Other: diff --git a/tests/test_sidecar.py b/tests/test_sidecar.py index c109f63..1b94e54 100644 --- a/tests/test_sidecar.py +++ b/tests/test_sidecar.py @@ -56,6 +56,40 @@ def __snapclass_ready__(self, *, snapshot): assert observed == ["# Dusk Court\n"] +def test_sidecar_assignment_in_loaded_hook_does_not_rewrite_files(tmp_path): + articles = Stash(tmp_path / "world") / "article" + hook_calls: list[str] = [] + + @snapclass("{self.slug}/article.yml", stash=articles, manual=True) + class Article: + slug: str + content_file: str = "" + body: str = sidecar.text(field="content_file", default="{self.slug}.md") + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied and sidecar hook writes stay in memory.""" + if not hook_calls: + self.body = "Hook body\n" + hook_calls.append("loaded") + + metadata = tmp_path / "world" / "article" / "dusk-court" / "article.yml" + body = tmp_path / "world" / "article" / "dusk-court" / "dusk-court.md" + metadata.parent.mkdir(parents=True) + metadata.write_text("content_file: dusk-court.md\n", encoding="utf-8") + body.write_text("File body\n", encoding="utf-8") + + article = Article.snapshots.get("dusk-court") + + assert article.body == "Hook body\n" + assert body.read_text(encoding="utf-8") == "File body\n" + assert metadata.read_text(encoding="utf-8") == "content_file: dusk-court.md\n" + + article.snapshot.load() + + assert article.body == "File body\n" + assert hook_calls == ["loaded", "loaded"] + + def test_text_sidecar_can_use_explicit_stash(tmp_path): app = Stash(tmp_path / "world") articles = app / "article" From 8fb3705c4b0dc65375b4d92c336d9fc661c3ade5 Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 09:46:59 -0500 Subject: [PATCH 04/11] persist suppressed sidecar overrides on save --- src/snapclass/schemas.py | 1 + src/snapclass/sidecar.py | 16 ++++++++++++ tests/test_sidecar.py | 55 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 72 insertions(+) diff --git a/src/snapclass/schemas.py b/src/snapclass/schemas.py index 765cc6a..3d6e3ec 100644 --- a/src/snapclass/schemas.py +++ b/src/snapclass/schemas.py @@ -836,6 +836,7 @@ def save( with _write_lock_for(current_path): self._check_write_conflict(current_path) sidecar.reconcile_before_save(self._instance, current_path) + sidecar.flush_overrides(self._instance) data = _to_data( self._instance, self._config, diff --git a/src/snapclass/sidecar.py b/src/snapclass/sidecar.py index 32a6308..3c44bc4 100644 --- a/src/snapclass/sidecar.py +++ b/src/snapclass/sidecar.py @@ -152,6 +152,22 @@ def clear_overrides(instance: object) -> None: object.__setattr__(instance, _OVERRIDES_ATTR, {}) +def flush_overrides(instance: object) -> None: + """Write suppressed sidecar assignments as part of an enclosing save.""" + overrides = dict(getattr(instance, _OVERRIDES_ATTR, {})) + if not overrides: + return + descriptors = { + getattr(descriptor, "_name", None): descriptor + for descriptor in _descriptors_for(type(instance)) + } + for name, value in overrides.items(): + descriptor = descriptors.get(name) + if descriptor is not None: + descriptor.snapshot(instance).write(value, save_metadata=False) + descriptor._clear_override(instance) + + class SidecarText(str): snapshot: "SidecarSnapshot" diff --git a/tests/test_sidecar.py b/tests/test_sidecar.py index 1b94e54..1864474 100644 --- a/tests/test_sidecar.py +++ b/tests/test_sidecar.py @@ -90,6 +90,61 @@ def __snapclass_loaded__(self, *, snapshot, path): assert hook_calls == ["loaded", "loaded"] +def test_ready_hook_sidecar_assignment_persists_on_initial_auto_save(tmp_path): + articles = Stash(tmp_path / "world") / "article" + + @snapclass("{self.slug}/article.yml", stash=articles) + class Article: + slug: str + content_file: str = "" + body: str = sidecar.text(field="content_file", default="{self.slug}.md") + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and initial sidecar state can be prepared.""" + self.body = "Ready body\n" + + article = Article("dusk-court") + + metadata = tmp_path / "world" / "article" / "dusk-court" / "article.yml" + body = tmp_path / "world" / "article" / "dusk-court" / "dusk-court.md" + + assert article.body == "Ready body\n" + assert body.read_text(encoding="utf-8") == "Ready body\n" + assert metadata.read_text(encoding="utf-8") == "content_file: dusk-court.md\n" + assert Article.snapshots.get("dusk-court").body == "Ready body\n" + + +def test_loaded_hook_sidecar_assignment_persists_on_explicit_save(tmp_path): + articles = Stash(tmp_path / "world") / "article" + hook_calls: list[str] = [] + + @snapclass("{self.slug}/article.yml", stash=articles, manual=True) + class Article: + slug: str + content_file: str = "" + body: str = sidecar.text(field="content_file", default="{self.slug}.md") + + def __snapclass_loaded__(self, *, snapshot, path): + """File data is applied before sidecar changes are explicitly saved.""" + if not hook_calls: + self.body = "Hook body\n" + hook_calls.append("loaded") + + metadata = tmp_path / "world" / "article" / "dusk-court" / "article.yml" + body = tmp_path / "world" / "article" / "dusk-court" / "dusk-court.md" + metadata.parent.mkdir(parents=True) + metadata.write_text("content_file: dusk-court.md\n", encoding="utf-8") + body.write_text("File body\n", encoding="utf-8") + + article = Article.snapshots.get("dusk-court") + article.snapshot.save() + reloaded = Article.snapshots.get("dusk-court") + + assert body.read_text(encoding="utf-8") == "Hook body\n" + assert metadata.read_text(encoding="utf-8") == "content_file: dusk-court.md\n" + assert reloaded.body == "Hook body\n" + + def test_text_sidecar_can_use_explicit_stash(tmp_path): app = Stash(tmp_path / "world") articles = app / "article" From 1bcbfc1c28289f94bde74832fc01fba6d34e4cec Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 10:28:08 -0500 Subject: [PATCH 05/11] document snapshot lifecycle hooks in fluency skill --- skills/snapclass-fluency/SKILL.md | 37 +++++++++++++++++++ .../skills/snapclass-fluency/SKILL.md | 37 +++++++++++++++++++ 2 files changed, 74 insertions(+) diff --git a/skills/snapclass-fluency/SKILL.md b/skills/snapclass-fluency/SKILL.md index d3d0b72..bb56c6b 100644 --- a/skills/snapclass-fluency/SKILL.md +++ b/skills/snapclass-fluency/SKILL.md @@ -880,6 +880,39 @@ By default, snapclass hooks can save changed instances automatically according t Use `manual=True` for models where mutation should stay in memory until saved. +Use lifecycle hooks when a model needs transient runtime state around snapshot +attachment and file loads: + +```python +@snapclass("{self.name}.yml") +class Chat: + name: str + runtime: object | None = None + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and the object is usable.""" + if self.runtime is None: + self.runtime = build_runtime_defaults() + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied to the object.""" + self.runtime = rebuild_runtime_from_loaded_state(self) +``` + +`__snapclass_ready__(self, *, snapshot)` runs once per attached snapshot after +the object is usable. On existing-file loads, file data may already be applied +before `ready` runs, so `ready` should establish missing live defaults rather +than overwrite persisted fields. + +`__snapclass_loaded__(self, *, snapshot, path)` runs after each successful +`Snapshot.load()`, including `obj.load()` and `snapshot.text = ...`. Use it for +authoritative post-load rebuilds from YAML/JSON/TOML/text state. + +Lifecycle hooks run with automatic snapclass save/reload behavior suppressed +for that instance. Hook mutations should be idempotent. Sidecar assignments made +inside a lifecycle hook stay in memory while the hook runs; a later +`snapshot.save()` persists them as part of that save. + Use `frozen(...)` or `hooks.disabled(...)` to temporarily suspend automatic saves: ```python @@ -1046,6 +1079,7 @@ Prefer targeted tests around the public story: - Serializer behavior for downstream-inspired cases. - Collection stash binding. - Unknown data, migration, defaults, minimal diffs, write delay, and conflict behavior. +- Lifecycle hook ordering, idempotency, autosave suppression, sidecar behavior, and file-backed reloads. For sidecars, assert both value behavior and snapshot behavior: @@ -1096,6 +1130,9 @@ Keep code comments rare and useful. A small comment is good before tricky descri - Sidecar values should act like `str` or `bytes`, with `.snapshot` for file details. - Sidecars default to the model's stash; explicit sidecar stashes can override or compose under it. - Sidecar fields should stay out of YAML and `dataclasses.fields(...)`. +- `__snapclass_ready__` should be one-shot per attached snapshot; `__snapclass_loaded__` can run on every load. +- Lifecycle hook mutations should not trigger automatic saves or reload loops. +- Sidecar values assigned inside lifecycle hooks should persist only through an enclosing or later explicit save. - `Fresh.List`, `Fresh.Dict`, and friends must produce fresh field objects and fresh values. - `defaults=True` controls whether default-valued fields are written. - `field(default_factory=dict)` is the standard dataclass spell; `Fresh.Dict` is the snapclass-friendly shorthand. diff --git a/src/snapclass/skills/snapclass-fluency/SKILL.md b/src/snapclass/skills/snapclass-fluency/SKILL.md index d3d0b72..bb56c6b 100644 --- a/src/snapclass/skills/snapclass-fluency/SKILL.md +++ b/src/snapclass/skills/snapclass-fluency/SKILL.md @@ -880,6 +880,39 @@ By default, snapclass hooks can save changed instances automatically according t Use `manual=True` for models where mutation should stay in memory until saved. +Use lifecycle hooks when a model needs transient runtime state around snapshot +attachment and file loads: + +```python +@snapclass("{self.name}.yml") +class Chat: + name: str + runtime: object | None = None + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and the object is usable.""" + if self.runtime is None: + self.runtime = build_runtime_defaults() + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied to the object.""" + self.runtime = rebuild_runtime_from_loaded_state(self) +``` + +`__snapclass_ready__(self, *, snapshot)` runs once per attached snapshot after +the object is usable. On existing-file loads, file data may already be applied +before `ready` runs, so `ready` should establish missing live defaults rather +than overwrite persisted fields. + +`__snapclass_loaded__(self, *, snapshot, path)` runs after each successful +`Snapshot.load()`, including `obj.load()` and `snapshot.text = ...`. Use it for +authoritative post-load rebuilds from YAML/JSON/TOML/text state. + +Lifecycle hooks run with automatic snapclass save/reload behavior suppressed +for that instance. Hook mutations should be idempotent. Sidecar assignments made +inside a lifecycle hook stay in memory while the hook runs; a later +`snapshot.save()` persists them as part of that save. + Use `frozen(...)` or `hooks.disabled(...)` to temporarily suspend automatic saves: ```python @@ -1046,6 +1079,7 @@ Prefer targeted tests around the public story: - Serializer behavior for downstream-inspired cases. - Collection stash binding. - Unknown data, migration, defaults, minimal diffs, write delay, and conflict behavior. +- Lifecycle hook ordering, idempotency, autosave suppression, sidecar behavior, and file-backed reloads. For sidecars, assert both value behavior and snapshot behavior: @@ -1096,6 +1130,9 @@ Keep code comments rare and useful. A small comment is good before tricky descri - Sidecar values should act like `str` or `bytes`, with `.snapshot` for file details. - Sidecars default to the model's stash; explicit sidecar stashes can override or compose under it. - Sidecar fields should stay out of YAML and `dataclasses.fields(...)`. +- `__snapclass_ready__` should be one-shot per attached snapshot; `__snapclass_loaded__` can run on every load. +- Lifecycle hook mutations should not trigger automatic saves or reload loops. +- Sidecar values assigned inside lifecycle hooks should persist only through an enclosing or later explicit save. - `Fresh.List`, `Fresh.Dict`, and friends must produce fresh field objects and fresh values. - `defaults=True` controls whether default-valued fields are written. - `field(default_factory=dict)` is the standard dataclass spell; `Fresh.Dict` is the snapclass-friendly shorthand. From b51f26d40e2f0646e6f458a8b2c2b9fb5a114528 Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 10:43:39 -0500 Subject: [PATCH 06/11] recheck lifecycle-mutated snapshot paths --- src/snapclass/schemas.py | 11 ++++++++--- src/snapclass/sidecar.py | 32 ++++++++++++++++++++++++++++---- tests/test_behavior_contracts.py | 25 +++++++++++++++++++++++++ tests/test_sidecar.py | 24 ++++++++++++++++++++++++ 4 files changed, 85 insertions(+), 7 deletions(-) diff --git a/src/snapclass/schemas.py b/src/snapclass/schemas.py index 3d6e3ec..518eab2 100644 --- a/src/snapclass/schemas.py +++ b/src/snapclass/schemas.py @@ -530,12 +530,16 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: _apply_sidecar_values(self, pending_sidecars, save_metadata=False) _apply_sidecar_values(self, sidecar_values, save_metadata=False) object.__setattr__(self, "_snapclass_initializing", False) - if _auto_enabled(config, self) and self.snapshot.exists: + automatic = _auto_enabled(config, self) + if automatic and self.snapshot.exists: self.snapshot.load(_initial=True) else: _mark_snapshot_ready(self) - if _auto_enabled(config, self): - self.snapshot.save() + if automatic: + if self.snapshot.exists: + self.snapshot.load() + else: + self.snapshot.save() cls.__init__ = __init__ @@ -832,6 +836,7 @@ def save( __tracebackhide__ = sessions.HIDDEN_TRACEBACK if path is not None: self.path = path + sidecar.prepare_overrides(self._instance) current_path = self._require_path() with _write_lock_for(current_path): self._check_write_conflict(current_path) diff --git a/src/snapclass/sidecar.py b/src/snapclass/sidecar.py index 3c44bc4..ae79d0c 100644 --- a/src/snapclass/sidecar.py +++ b/src/snapclass/sidecar.py @@ -146,21 +146,37 @@ def _override_value(self, instance: object) -> object: return _MISSING return getattr(instance, _OVERRIDES_ATTR, {}).get(name, _MISSING) + def _prepare_pointer(self, instance: object) -> None: + """Apply the pointer-field update that a later sidecar write would make.""" + if not self.field: + return + relative = self.snapshot(instance).relative_path + object.__setattr__(instance, self.field, relative.as_posix()) + def clear_overrides(instance: object) -> None: """Clear all suppressed sidecar assignments for an instance.""" object.__setattr__(instance, _OVERRIDES_ATTR, {}) +def prepare_overrides(instance: object) -> None: + """Apply sidecar pointer fields before resolving the enclosing snapshot path.""" + overrides = dict(getattr(instance, _OVERRIDES_ATTR, {})) + if not overrides: + return + descriptors = _override_descriptors(instance) + for name in overrides: + descriptor = descriptors.get(name) + if descriptor is not None: + descriptor._prepare_pointer(instance) + + def flush_overrides(instance: object) -> None: """Write suppressed sidecar assignments as part of an enclosing save.""" overrides = dict(getattr(instance, _OVERRIDES_ATTR, {})) if not overrides: return - descriptors = { - getattr(descriptor, "_name", None): descriptor - for descriptor in _descriptors_for(type(instance)) - } + descriptors = _override_descriptors(instance) for name, value in overrides.items(): descriptor = descriptors.get(name) if descriptor is not None: @@ -168,6 +184,14 @@ def flush_overrides(instance: object) -> None: descriptor._clear_override(instance) +def _override_descriptors(instance: object) -> dict[str | None, SidecarDescriptor]: + """Return sidecar descriptors keyed by their owning model attribute name.""" + return { + getattr(descriptor, "_name", None): descriptor + for descriptor in _descriptors_for(type(instance)) + } + + class SidecarText(str): snapshot: "SidecarSnapshot" diff --git a/tests/test_behavior_contracts.py b/tests/test_behavior_contracts.py index c0b51b8..34b9295 100644 --- a/tests/test_behavior_contracts.py +++ b/tests/test_behavior_contracts.py @@ -272,6 +272,31 @@ def __snapclass_ready__(self, *, snapshot): assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" +def test_ready_hook_path_change_loads_existing_snapshot_before_create(tmp_path): + @snapclass("{self.name}.yml", stash=Stash(tmp_path)) + class Item: + name: str + value: str = "" + ready_seen: str = field(init=False) + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and may normalize path fields before create.""" + self.name = self.name.removeprefix("draft-") + self.ready_seen = self.value + if not self.value: + self.value = "ready" + + (tmp_path / "sample.yml").write_text("value: file\n", encoding="utf-8") + + item = Item("draft-sample") + + assert item.name == "sample" + assert item.ready_seen == "" + assert item.value == "file" + assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" + assert not (tmp_path / "draft-sample.yml").exists() + + def test_ready_hook_runs_once_but_loaded_runs_on_each_load(tmp_path): calls: list[str] = [] diff --git a/tests/test_sidecar.py b/tests/test_sidecar.py index 1864474..04811d9 100644 --- a/tests/test_sidecar.py +++ b/tests/test_sidecar.py @@ -145,6 +145,30 @@ def __snapclass_loaded__(self, *, snapshot, path): assert reloaded.body == "Hook body\n" +def test_ready_hook_sidecar_pointer_retargets_snapshot_save_path(tmp_path): + @snapclass("{self.content_file}.yml", stash=Stash(tmp_path), manual=True) + class Article: + name: str + content_file: str = "" + body: str = sidecar.text(field="content_file", default="{self.name}.md") + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and sidecar pointers can retarget first save.""" + self.body = "Ready body\n" + + article = Article("dusk") + + article.snapshot.save() + + metadata = tmp_path / "dusk.md.yml" + body = tmp_path / "dusk.md" + + assert article.content_file == "dusk.md" + assert article.snapshot.path == metadata + assert body.read_text(encoding="utf-8") == "Ready body\n" + assert metadata.read_text(encoding="utf-8") == "name: dusk\n" + + def test_text_sidecar_can_use_explicit_stash(tmp_path): app = Stash(tmp_path / "world") articles = app / "article" From dd3e1a8af87b83cc47088abfe8ee118415037775 Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 11:37:14 -0500 Subject: [PATCH 07/11] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/snapclass/sidecar.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/src/snapclass/sidecar.py b/src/snapclass/sidecar.py index ae79d0c..db6f66f 100644 --- a/src/snapclass/sidecar.py +++ b/src/snapclass/sidecar.py @@ -184,12 +184,14 @@ def flush_overrides(instance: object) -> None: descriptor._clear_override(instance) -def _override_descriptors(instance: object) -> dict[str | None, SidecarDescriptor]: +def _override_descriptors(instance: object) -> dict[str, SidecarDescriptor]: """Return sidecar descriptors keyed by their owning model attribute name.""" - return { - getattr(descriptor, "_name", None): descriptor - for descriptor in _descriptors_for(type(instance)) - } + descriptors: dict[str, SidecarDescriptor] = {} + for descriptor in _descriptors_for(type(instance)): + name = getattr(descriptor, "_name", None) + if name is not None: + descriptors[name] = descriptor + return descriptors class SidecarText(str): From 8ffc7a4458052579187a1c5504f7f2b03da6a313 Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 11:41:07 -0500 Subject: [PATCH 08/11] recheck collection ready paths --- src/snapclass/collections.py | 10 ++++++++-- tests/test_behavior_contracts.py | 25 +++++++++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/src/snapclass/collections.py b/src/snapclass/collections.py index f6d3e6f..a1c3ec5 100644 --- a/src/snapclass/collections.py +++ b/src/snapclass/collections.py @@ -48,11 +48,17 @@ def get_or_create(self, *args: Any, **kwargs: Any) -> Any: __tracebackhide__ = sessions.HIDDEN_TRACEBACK instance = self._empty_instance(*args, **kwargs, include_defaults=True) _attach_snapshot(instance, self.model.__snapclass_config__, self._stash) - with _write_lock_for(instance.snapshot._require_path()): + initial_path = instance.snapshot._require_path() + with _write_lock_for(initial_path): if instance.snapshot.exists: instance.snapshot.load(_initial=True) + return instance + _mark_snapshot_ready(instance) + current_path = instance.snapshot._require_path() + with _write_lock_for(current_path): + if instance.snapshot.exists: + instance.snapshot.load() else: - _mark_snapshot_ready(instance) instance.snapshot.save() return instance diff --git a/tests/test_behavior_contracts.py b/tests/test_behavior_contracts.py index 34b9295..dd01be9 100644 --- a/tests/test_behavior_contracts.py +++ b/tests/test_behavior_contracts.py @@ -297,6 +297,31 @@ def __snapclass_ready__(self, *, snapshot): assert not (tmp_path / "draft-sample.yml").exists() +def test_get_or_create_rechecks_ready_changed_path_before_create(tmp_path): + @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) + class Item: + name: str + value: str = "" + ready_seen: str = field(init=False) + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and may normalize collection lookup fields.""" + self.name = self.name.removeprefix("draft-") + self.ready_seen = self.value + if not self.value: + self.value = "ready" + + (tmp_path / "sample.yml").write_text("value: file\n", encoding="utf-8") + + item = Item.snapshots.get_or_create("draft-sample") + + assert item.name == "sample" + assert item.ready_seen == "" + assert item.value == "file" + assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" + assert not (tmp_path / "draft-sample.yml").exists() + + def test_ready_hook_runs_once_but_loaded_runs_on_each_load(tmp_path): calls: list[str] = [] From 463af9d62aaa4c42dc9f97ac1ce842ce1b28f4e8 Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 12:24:03 -0500 Subject: [PATCH 09/11] follow lifecycle-retargeted load paths --- src/snapclass/schemas.py | 21 ++++++++++++ tests/test_behavior_contracts.py | 56 ++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/src/snapclass/schemas.py b/src/snapclass/schemas.py index 518eab2..a454fbc 100644 --- a/src/snapclass/schemas.py +++ b/src/snapclass/schemas.py @@ -866,11 +866,15 @@ def load( path: str | os.PathLike[str] | None = None, *, _initial: bool = False, + _visited_paths: set[Path] | None = None, ) -> None: __tracebackhide__ = sessions.HIDDEN_TRACEBACK if path is not None: self.path = path current_path = self._require_path() + if _visited_paths is None: + _visited_paths = set() + _visited_paths.add(current_path) text = current_path.read_text(encoding="utf-8") try: data = _load_data(current_path, text, self._config, self.stash) @@ -890,9 +894,26 @@ def load( self.modified = False _mark_snapshot_ready(self._instance) _mark_snapshot_loaded(self._instance, current_path) + self._reload_after_lifecycle_path_change(current_path, _visited_paths) finally: object.__setattr__(self._instance, "_snapclass_loading", False) + def _reload_after_lifecycle_path_change( + self, + loaded_path: Path, + visited_paths: set[Path], + ) -> None: + """Load an existing post-hook path before returning from a retargeted load.""" + current_path = self._require_path() + if current_path == loaded_path or not current_path.exists(): + return + if current_path in visited_paths: + raise SnapclassError( + "Lifecycle hooks changed snapshot path in a load cycle: " + f"{loaded_path} -> {current_path}" + ) + self.load(_visited_paths=visited_paths) + def _require_path(self) -> Path: try: path = self.path diff --git a/tests/test_behavior_contracts.py b/tests/test_behavior_contracts.py index dd01be9..db7ab10 100644 --- a/tests/test_behavior_contracts.py +++ b/tests/test_behavior_contracts.py @@ -322,6 +322,62 @@ def __snapclass_ready__(self, *, snapshot): assert not (tmp_path / "draft-sample.yml").exists() +def test_load_follows_existing_ready_changed_path_before_save(tmp_path): + loaded_paths: list[str] = [] + + @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) + class Item: + name: str + value: str = "" + + def __snapclass_ready__(self, *, snapshot): + """Snapshot is attached and may normalize loaded path fields.""" + self.name = self.name.removeprefix("draft-") + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied before any redirected load is followed.""" + loaded_paths.append(path.name) + + (tmp_path / "draft-sample.yml").write_text("value: draft\n", encoding="utf-8") + (tmp_path / "sample.yml").write_text("value: file\n", encoding="utf-8") + + item = Item.snapshots.get("draft-sample") + + assert item.name == "sample" + assert item.value == "file" + assert loaded_paths == ["draft-sample.yml", "sample.yml"] + item.snapshot.save() + assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" + assert (tmp_path / "draft-sample.yml").read_text(encoding="utf-8") == "value: draft\n" + + +def test_load_follows_existing_loaded_changed_path_before_save(tmp_path): + loaded_paths: list[str] = [] + + @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) + class Item: + name: str + value: str = "" + + def __snapclass_loaded__(self, *, snapshot, path): + """File data has been applied and may retarget the loaded snapshot.""" + loaded_paths.append(path.name) + if path.name == "draft-sample.yml": + self.name = "sample" + + (tmp_path / "draft-sample.yml").write_text("value: draft\n", encoding="utf-8") + (tmp_path / "sample.yml").write_text("value: file\n", encoding="utf-8") + + item = Item.snapshots.get("draft-sample") + + assert item.name == "sample" + assert item.value == "file" + assert loaded_paths == ["draft-sample.yml", "sample.yml"] + item.snapshot.save() + assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" + assert (tmp_path / "draft-sample.yml").read_text(encoding="utf-8") == "value: draft\n" + + def test_ready_hook_runs_once_but_loaded_runs_on_each_load(tmp_path): calls: list[str] = [] From f338609fbd2441368a7c4cb63ef7cb465c8a5b20 Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 13:01:54 -0500 Subject: [PATCH 10/11] Trim lifecycle hooks to core contract --- skills/snapclass-fluency/SKILL.md | 9 +- src/snapclass/collections.py | 12 +- src/snapclass/schemas.py | 73 ++++++----- src/snapclass/sidecar.py | 93 -------------- .../skills/snapclass-fluency/SKILL.md | 9 +- tests/test_behavior_contracts.py | 119 +++++------------- tests/test_sidecar.py | 106 +--------------- 7 files changed, 94 insertions(+), 327 deletions(-) diff --git a/skills/snapclass-fluency/SKILL.md b/skills/snapclass-fluency/SKILL.md index bb56c6b..d7b7f41 100644 --- a/skills/snapclass-fluency/SKILL.md +++ b/skills/snapclass-fluency/SKILL.md @@ -909,9 +909,9 @@ than overwrite persisted fields. authoritative post-load rebuilds from YAML/JSON/TOML/text state. Lifecycle hooks run with automatic snapclass save/reload behavior suppressed -for that instance. Hook mutations should be idempotent. Sidecar assignments made -inside a lifecycle hook stay in memory while the hook runs; a later -`snapshot.save()` persists them as part of that save. +for that instance. Hook mutations should be idempotent and must not change +fields used by the snapshot pattern. Sidecars may be read inside hooks, but +sidecar writes should happen after the hook has returned. Use `frozen(...)` or `hooks.disabled(...)` to temporarily suspend automatic saves: @@ -1132,7 +1132,8 @@ Keep code comments rare and useful. A small comment is good before tricky descri - Sidecar fields should stay out of YAML and `dataclasses.fields(...)`. - `__snapclass_ready__` should be one-shot per attached snapshot; `__snapclass_loaded__` can run on every load. - Lifecycle hook mutations should not trigger automatic saves or reload loops. -- Sidecar values assigned inside lifecycle hooks should persist only through an enclosing or later explicit save. +- Lifecycle hooks must not change fields used by the snapshot pattern; normalize those values before loading or creating snapshots. +- Sidecars may be read inside lifecycle hooks; write sidecars after hooks return. - `Fresh.List`, `Fresh.Dict`, and friends must produce fresh field objects and fresh values. - `defaults=True` controls whether default-valued fields are written. - `field(default_factory=dict)` is the standard dataclass spell; `Fresh.Dict` is the snapclass-friendly shorthand. diff --git a/src/snapclass/collections.py b/src/snapclass/collections.py index a1c3ec5..8ae7145 100644 --- a/src/snapclass/collections.py +++ b/src/snapclass/collections.py @@ -52,14 +52,12 @@ def get_or_create(self, *args: Any, **kwargs: Any) -> Any: with _write_lock_for(initial_path): if instance.snapshot.exists: instance.snapshot.load(_initial=True) - return instance - _mark_snapshot_ready(instance) - current_path = instance.snapshot._require_path() - with _write_lock_for(current_path): - if instance.snapshot.exists: - instance.snapshot.load() else: - instance.snapshot.save() + _mark_snapshot_ready(instance) + if instance.snapshot.exists: + instance.snapshot.load() + else: + instance.snapshot.save() return instance def all(self, *, _exclude: str = "") -> Iterator[Any]: diff --git a/src/snapclass/schemas.py b/src/snapclass/schemas.py index a454fbc..5dc488f 100644 --- a/src/snapclass/schemas.py +++ b/src/snapclass/schemas.py @@ -436,13 +436,19 @@ def _mark_snapshot_ready(instance: object) -> None: return if getattr(snapshot, "_ready", False): return + hook = getattr(instance, "__snapclass_ready__", None) + expected_path = snapshot.path if hook is not None else None snapshot._ready = True try: _call_snapshot_lifecycle_hook( instance, "__snapclass_ready__", snapshot=snapshot, - path=_snapshot_path_or_none(snapshot), + ) + _ensure_snapshot_path_unchanged( + snapshot, + expected_path, + "__snapclass_ready__", ) except Exception: snapshot._ready = False @@ -454,12 +460,19 @@ def _mark_snapshot_loaded(instance: object, path: Path) -> None: snapshot = getattr(instance, "snapshot", None) if snapshot is None: return + hook = getattr(instance, "__snapclass_loaded__", None) + expected_path = snapshot.path if hook is not None else None _call_snapshot_lifecycle_hook( instance, "__snapclass_loaded__", snapshot=snapshot, path=path, ) + _ensure_snapshot_path_unchanged( + snapshot, + expected_path, + "__snapclass_loaded__", + ) def _call_snapshot_lifecycle_hook( @@ -501,12 +514,29 @@ def _snapclass_hook_context(instance: object) -> Iterator[None]: object.__setattr__(instance, "_snapclass_loading", previous_loading) -def _snapshot_path_or_none(snapshot: "Snapshot") -> Path | None: - """Return the resolved snapshot path when it can be used for diagnostics.""" +def _ensure_snapshot_path_unchanged( + snapshot: "Snapshot", + expected_path: Path | None, + hook_name: str, +) -> None: + """Raise when a lifecycle hook retargets the snapshot path.""" + if expected_path is None: + return try: - return snapshot.path - except Exception: - return None + current_path = snapshot._require_path() + except Exception as exc: + raise SnapclassError( + f"{hook_name} left snapshot path unresolved after lifecycle hook; " + "normalize snapshot path fields before loading or creating " + f"snapshots: {exc}" + ) from exc + if current_path == expected_path: + return + raise SnapclassError( + f"{hook_name} changed snapshot path from {expected_path} to " + f"{current_path}; normalize snapshot path fields before loading or " + "creating snapshots" + ) def _install(cls: type, config: Config) -> None: @@ -556,9 +586,10 @@ def __setattr__(self, name: str, value: Any) -> None: object.__setattr__(self, _PENDING_SIDECARS_ATTR, pending) return if getattr(self, "_snapclass_hooks_suppressed", False): - sidecar_descriptor._set_override(self, value) - return - sidecar_descriptor._clear_override(self) + raise SnapclassError( + f"Cannot assign sidecar {name!r} during snapclass lifecycle " + "hooks; write sidecars after the hook has returned" + ) sidecar_descriptor.snapshot(self).write(value) return snapshot = getattr(self, "snapshot", None) @@ -836,12 +867,10 @@ def save( __tracebackhide__ = sessions.HIDDEN_TRACEBACK if path is not None: self.path = path - sidecar.prepare_overrides(self._instance) current_path = self._require_path() with _write_lock_for(current_path): self._check_write_conflict(current_path) sidecar.reconcile_before_save(self._instance, current_path) - sidecar.flush_overrides(self._instance) data = _to_data( self._instance, self._config, @@ -866,15 +895,11 @@ def load( path: str | os.PathLike[str] | None = None, *, _initial: bool = False, - _visited_paths: set[Path] | None = None, ) -> None: __tracebackhide__ = sessions.HIDDEN_TRACEBACK if path is not None: self.path = path current_path = self._require_path() - if _visited_paths is None: - _visited_paths = set() - _visited_paths.add(current_path) text = current_path.read_text(encoding="utf-8") try: data = _load_data(current_path, text, self._config, self.stash) @@ -883,7 +908,6 @@ def load( object.__setattr__(self._instance, "_snapclass_loading", True) try: try: - sidecar.clear_overrides(self._instance) _apply_data(self._instance, data, preserve_non_default=_initial) except _CoercionError as exc: raise SnapclassError(_schema_mismatch_message(current_path, exc)) from exc @@ -894,26 +918,9 @@ def load( self.modified = False _mark_snapshot_ready(self._instance) _mark_snapshot_loaded(self._instance, current_path) - self._reload_after_lifecycle_path_change(current_path, _visited_paths) finally: object.__setattr__(self._instance, "_snapclass_loading", False) - def _reload_after_lifecycle_path_change( - self, - loaded_path: Path, - visited_paths: set[Path], - ) -> None: - """Load an existing post-hook path before returning from a retargeted load.""" - current_path = self._require_path() - if current_path == loaded_path or not current_path.exists(): - return - if current_path in visited_paths: - raise SnapclassError( - "Lifecycle hooks changed snapshot path in a load cycle: " - f"{loaded_path} -> {current_path}" - ) - self.load(_visited_paths=visited_paths) - def _require_path(self) -> Path: try: path = self.path diff --git a/src/snapclass/sidecar.py b/src/snapclass/sidecar.py index db6f66f..b1bc217 100644 --- a/src/snapclass/sidecar.py +++ b/src/snapclass/sidecar.py @@ -12,7 +12,6 @@ _Kind = Literal["text", "bytes"] _StashLike = Stash | str | os.PathLike[str] _MISSING = object() -_OVERRIDES_ATTR = "__snapclass_sidecar_overrides__" class SidecarMissingError(FileNotFoundError): @@ -84,10 +83,6 @@ class SidecarDescriptor: encoding: str = "utf-8" stash: Stash | None = None - def __set_name__(self, owner: type, name: str) -> None: - """Remember the model attribute name for suppressed in-memory values.""" - object.__setattr__(self, "_name", name) - def __get__(self, instance: object | None, owner: type | None = None): if instance is None: return self @@ -101,98 +96,10 @@ def snapshot(self, instance: object) -> "SidecarSnapshot": def value(self, instance: object) -> "SidecarText | SidecarBytes": snapshot = self.snapshot(instance) - override = self._override_value(instance) - if override is not _MISSING: - if self.kind == "text": - return SidecarText(cast(str, override), snapshot) - return SidecarBytes(cast(builtins.bytes, override), snapshot) if self.kind == "text": return SidecarText(snapshot.read(default=""), snapshot) return SidecarBytes(snapshot.read(default=b""), snapshot) - def _set_override(self, instance: object, value: str | builtins.bytes) -> None: - """Store a sidecar assignment in memory without writing sidecar files.""" - if self.kind == "text" and not isinstance(value, str): - raise TypeError("Text sidecars require str values") - if self.kind == "bytes" and not isinstance( - value, - (builtins.bytes, bytearray, memoryview), - ): - raise TypeError("Bytes sidecars require bytes-like values") - name = getattr(self, "_name", None) - if name is None: - return - overrides = dict(getattr(instance, _OVERRIDES_ATTR, {})) - overrides[name] = ( - builtins.bytes(value) if self.kind == "bytes" else value - ) - object.__setattr__(instance, _OVERRIDES_ATTR, overrides) - - def _clear_override(self, instance: object) -> None: - """Remove an in-memory sidecar assignment for this descriptor.""" - name = getattr(self, "_name", None) - if name is None: - return - overrides = dict(getattr(instance, _OVERRIDES_ATTR, {})) - if name not in overrides: - return - del overrides[name] - object.__setattr__(instance, _OVERRIDES_ATTR, overrides) - - def _override_value(self, instance: object) -> object: - """Return a suppressed in-memory value or the missing sentinel.""" - name = getattr(self, "_name", None) - if name is None: - return _MISSING - return getattr(instance, _OVERRIDES_ATTR, {}).get(name, _MISSING) - - def _prepare_pointer(self, instance: object) -> None: - """Apply the pointer-field update that a later sidecar write would make.""" - if not self.field: - return - relative = self.snapshot(instance).relative_path - object.__setattr__(instance, self.field, relative.as_posix()) - - -def clear_overrides(instance: object) -> None: - """Clear all suppressed sidecar assignments for an instance.""" - object.__setattr__(instance, _OVERRIDES_ATTR, {}) - - -def prepare_overrides(instance: object) -> None: - """Apply sidecar pointer fields before resolving the enclosing snapshot path.""" - overrides = dict(getattr(instance, _OVERRIDES_ATTR, {})) - if not overrides: - return - descriptors = _override_descriptors(instance) - for name in overrides: - descriptor = descriptors.get(name) - if descriptor is not None: - descriptor._prepare_pointer(instance) - - -def flush_overrides(instance: object) -> None: - """Write suppressed sidecar assignments as part of an enclosing save.""" - overrides = dict(getattr(instance, _OVERRIDES_ATTR, {})) - if not overrides: - return - descriptors = _override_descriptors(instance) - for name, value in overrides.items(): - descriptor = descriptors.get(name) - if descriptor is not None: - descriptor.snapshot(instance).write(value, save_metadata=False) - descriptor._clear_override(instance) - - -def _override_descriptors(instance: object) -> dict[str, SidecarDescriptor]: - """Return sidecar descriptors keyed by their owning model attribute name.""" - descriptors: dict[str, SidecarDescriptor] = {} - for descriptor in _descriptors_for(type(instance)): - name = getattr(descriptor, "_name", None) - if name is not None: - descriptors[name] = descriptor - return descriptors - class SidecarText(str): snapshot: "SidecarSnapshot" diff --git a/src/snapclass/skills/snapclass-fluency/SKILL.md b/src/snapclass/skills/snapclass-fluency/SKILL.md index bb56c6b..d7b7f41 100644 --- a/src/snapclass/skills/snapclass-fluency/SKILL.md +++ b/src/snapclass/skills/snapclass-fluency/SKILL.md @@ -909,9 +909,9 @@ than overwrite persisted fields. authoritative post-load rebuilds from YAML/JSON/TOML/text state. Lifecycle hooks run with automatic snapclass save/reload behavior suppressed -for that instance. Hook mutations should be idempotent. Sidecar assignments made -inside a lifecycle hook stay in memory while the hook runs; a later -`snapshot.save()` persists them as part of that save. +for that instance. Hook mutations should be idempotent and must not change +fields used by the snapshot pattern. Sidecars may be read inside hooks, but +sidecar writes should happen after the hook has returned. Use `frozen(...)` or `hooks.disabled(...)` to temporarily suspend automatic saves: @@ -1132,7 +1132,8 @@ Keep code comments rare and useful. A small comment is good before tricky descri - Sidecar fields should stay out of YAML and `dataclasses.fields(...)`. - `__snapclass_ready__` should be one-shot per attached snapshot; `__snapclass_loaded__` can run on every load. - Lifecycle hook mutations should not trigger automatic saves or reload loops. -- Sidecar values assigned inside lifecycle hooks should persist only through an enclosing or later explicit save. +- Lifecycle hooks must not change fields used by the snapshot pattern; normalize those values before loading or creating snapshots. +- Sidecars may be read inside lifecycle hooks; write sidecars after hooks return. - `Fresh.List`, `Fresh.Dict`, and friends must produce fresh field objects and fresh values. - `defaults=True` controls whether default-valued fields are written. - `field(default_factory=dict)` is the standard dataclass spell; `Fresh.Dict` is the snapclass-friendly shorthand. diff --git a/tests/test_behavior_contracts.py b/tests/test_behavior_contracts.py index db7ab10..191b4bd 100644 --- a/tests/test_behavior_contracts.py +++ b/tests/test_behavior_contracts.py @@ -272,110 +272,56 @@ def __snapclass_ready__(self, *, snapshot): assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" -def test_ready_hook_path_change_loads_existing_snapshot_before_create(tmp_path): - @snapclass("{self.name}.yml", stash=Stash(tmp_path)) - class Item: - name: str - value: str = "" - ready_seen: str = field(init=False) - - def __snapclass_ready__(self, *, snapshot): - """Snapshot is attached and may normalize path fields before create.""" - self.name = self.name.removeprefix("draft-") - self.ready_seen = self.value - if not self.value: - self.value = "ready" - - (tmp_path / "sample.yml").write_text("value: file\n", encoding="utf-8") - - item = Item("draft-sample") - - assert item.name == "sample" - assert item.ready_seen == "" - assert item.value == "file" - assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" - assert not (tmp_path / "draft-sample.yml").exists() - - -def test_get_or_create_rechecks_ready_changed_path_before_create(tmp_path): - @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) - class Item: +def test_lifecycle_hooks_cannot_retarget_snapshot_paths(tmp_path): + @snapclass("{self.name}.yml", stash=Stash(tmp_path / "created")) + class CreatedItem: name: str value: str = "" - ready_seen: str = field(init=False) def __snapclass_ready__(self, *, snapshot): - """Snapshot is attached and may normalize collection lookup fields.""" - self.name = self.name.removeprefix("draft-") - self.ready_seen = self.value - if not self.value: - self.value = "ready" - - (tmp_path / "sample.yml").write_text("value: file\n", encoding="utf-8") - - item = Item.snapshots.get_or_create("draft-sample") - - assert item.name == "sample" - assert item.ready_seen == "" - assert item.value == "file" - assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" - assert not (tmp_path / "draft-sample.yml").exists() + """Snapshot pattern fields must be stable during create hooks.""" + self.name = "sample" + with pytest.raises( + SnapclassError, + match="__snapclass_ready__.*changed snapshot path", + ): + CreatedItem("draft") -def test_load_follows_existing_ready_changed_path_before_save(tmp_path): - loaded_paths: list[str] = [] - - @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) - class Item: + @snapclass("{self.name}.yml", stash=Stash(tmp_path / "collection"), manual=True) + class CollectionItem: name: str value: str = "" def __snapclass_ready__(self, *, snapshot): - """Snapshot is attached and may normalize loaded path fields.""" - self.name = self.name.removeprefix("draft-") - - def __snapclass_loaded__(self, *, snapshot, path): - """File data has been applied before any redirected load is followed.""" - loaded_paths.append(path.name) - - (tmp_path / "draft-sample.yml").write_text("value: draft\n", encoding="utf-8") - (tmp_path / "sample.yml").write_text("value: file\n", encoding="utf-8") - - item = Item.snapshots.get("draft-sample") - - assert item.name == "sample" - assert item.value == "file" - assert loaded_paths == ["draft-sample.yml", "sample.yml"] - item.snapshot.save() - assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" - assert (tmp_path / "draft-sample.yml").read_text(encoding="utf-8") == "value: draft\n" + """Snapshot pattern fields must be stable during collection create hooks.""" + self.name = "sample" + with pytest.raises( + SnapclassError, + match="__snapclass_ready__.*changed snapshot path", + ): + CollectionItem.snapshots.get_or_create("draft") -def test_load_follows_existing_loaded_changed_path_before_save(tmp_path): - loaded_paths: list[str] = [] + loaded_stash = tmp_path / "loaded" + loaded_stash.mkdir() - @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) - class Item: + @snapclass("{self.name}.yml", stash=Stash(loaded_stash), manual=True) + class LoadedItem: name: str value: str = "" def __snapclass_loaded__(self, *, snapshot, path): - """File data has been applied and may retarget the loaded snapshot.""" - loaded_paths.append(path.name) - if path.name == "draft-sample.yml": - self.name = "sample" - - (tmp_path / "draft-sample.yml").write_text("value: draft\n", encoding="utf-8") - (tmp_path / "sample.yml").write_text("value: file\n", encoding="utf-8") + """Snapshot pattern fields must be stable during load hooks.""" + self.name = "sample" - item = Item.snapshots.get("draft-sample") + (loaded_stash / "draft.yml").write_text("value: draft\n", encoding="utf-8") - assert item.name == "sample" - assert item.value == "file" - assert loaded_paths == ["draft-sample.yml", "sample.yml"] - item.snapshot.save() - assert (tmp_path / "sample.yml").read_text(encoding="utf-8") == "value: file\n" - assert (tmp_path / "draft-sample.yml").read_text(encoding="utf-8") == "value: draft\n" + with pytest.raises( + SnapclassError, + match="__snapclass_loaded__.*changed snapshot path", + ): + LoadedItem.snapshots.get("draft") def test_ready_hook_runs_once_but_loaded_runs_on_each_load(tmp_path): @@ -450,7 +396,8 @@ def __snapclass_ready__(self, *, snapshot): ready_message = str(ready_error.value) assert "__snapclass_ready__" in ready_message - assert "ready.yml" in ready_message + assert "ReadyItem" in ready_message + assert "ready failed" in ready_message @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True) class LoadedItem: diff --git a/tests/test_sidecar.py b/tests/test_sidecar.py index 04811d9..bfdced0 100644 --- a/tests/test_sidecar.py +++ b/tests/test_sidecar.py @@ -56,117 +56,23 @@ def __snapclass_ready__(self, *, snapshot): assert observed == ["# Dusk Court\n"] -def test_sidecar_assignment_in_loaded_hook_does_not_rewrite_files(tmp_path): +def test_sidecar_assignment_in_lifecycle_hook_raises_without_writing(tmp_path): articles = Stash(tmp_path / "world") / "article" - hook_calls: list[str] = [] @snapclass("{self.slug}/article.yml", stash=articles, manual=True) class Article: slug: str - content_file: str = "" - body: str = sidecar.text(field="content_file", default="{self.slug}.md") - - def __snapclass_loaded__(self, *, snapshot, path): - """File data has been applied and sidecar hook writes stay in memory.""" - if not hook_calls: - self.body = "Hook body\n" - hook_calls.append("loaded") - - metadata = tmp_path / "world" / "article" / "dusk-court" / "article.yml" - body = tmp_path / "world" / "article" / "dusk-court" / "dusk-court.md" - metadata.parent.mkdir(parents=True) - metadata.write_text("content_file: dusk-court.md\n", encoding="utf-8") - body.write_text("File body\n", encoding="utf-8") - - article = Article.snapshots.get("dusk-court") - - assert article.body == "Hook body\n" - assert body.read_text(encoding="utf-8") == "File body\n" - assert metadata.read_text(encoding="utf-8") == "content_file: dusk-court.md\n" - - article.snapshot.load() - - assert article.body == "File body\n" - assert hook_calls == ["loaded", "loaded"] - - -def test_ready_hook_sidecar_assignment_persists_on_initial_auto_save(tmp_path): - articles = Stash(tmp_path / "world") / "article" - - @snapclass("{self.slug}/article.yml", stash=articles) - class Article: - slug: str - content_file: str = "" - body: str = sidecar.text(field="content_file", default="{self.slug}.md") + body: str = sidecar.text("{self.slug}.md") def __snapclass_ready__(self, *, snapshot): - """Snapshot is attached and initial sidecar state can be prepared.""" + """Snapshot hooks may read sidecars, but writes happen after hooks.""" self.body = "Ready body\n" - article = Article("dusk-court") - - metadata = tmp_path / "world" / "article" / "dusk-court" / "article.yml" - body = tmp_path / "world" / "article" / "dusk-court" / "dusk-court.md" - - assert article.body == "Ready body\n" - assert body.read_text(encoding="utf-8") == "Ready body\n" - assert metadata.read_text(encoding="utf-8") == "content_file: dusk-court.md\n" - assert Article.snapshots.get("dusk-court").body == "Ready body\n" - + with pytest.raises(SnapclassError, match="Cannot assign sidecar 'body'"): + Article("dusk-court") -def test_loaded_hook_sidecar_assignment_persists_on_explicit_save(tmp_path): - articles = Stash(tmp_path / "world") / "article" - hook_calls: list[str] = [] - - @snapclass("{self.slug}/article.yml", stash=articles, manual=True) - class Article: - slug: str - content_file: str = "" - body: str = sidecar.text(field="content_file", default="{self.slug}.md") - - def __snapclass_loaded__(self, *, snapshot, path): - """File data is applied before sidecar changes are explicitly saved.""" - if not hook_calls: - self.body = "Hook body\n" - hook_calls.append("loaded") - - metadata = tmp_path / "world" / "article" / "dusk-court" / "article.yml" body = tmp_path / "world" / "article" / "dusk-court" / "dusk-court.md" - metadata.parent.mkdir(parents=True) - metadata.write_text("content_file: dusk-court.md\n", encoding="utf-8") - body.write_text("File body\n", encoding="utf-8") - - article = Article.snapshots.get("dusk-court") - article.snapshot.save() - reloaded = Article.snapshots.get("dusk-court") - - assert body.read_text(encoding="utf-8") == "Hook body\n" - assert metadata.read_text(encoding="utf-8") == "content_file: dusk-court.md\n" - assert reloaded.body == "Hook body\n" - - -def test_ready_hook_sidecar_pointer_retargets_snapshot_save_path(tmp_path): - @snapclass("{self.content_file}.yml", stash=Stash(tmp_path), manual=True) - class Article: - name: str - content_file: str = "" - body: str = sidecar.text(field="content_file", default="{self.name}.md") - - def __snapclass_ready__(self, *, snapshot): - """Snapshot is attached and sidecar pointers can retarget first save.""" - self.body = "Ready body\n" - - article = Article("dusk") - - article.snapshot.save() - - metadata = tmp_path / "dusk.md.yml" - body = tmp_path / "dusk.md" - - assert article.content_file == "dusk.md" - assert article.snapshot.path == metadata - assert body.read_text(encoding="utf-8") == "Ready body\n" - assert metadata.read_text(encoding="utf-8") == "name: dusk\n" + assert not body.exists() def test_text_sidecar_can_use_explicit_stash(tmp_path): From e577a01ccee99dd4c18f3ce0fa43efb6b46df4d7 Mon Sep 17 00:00:00 2001 From: Mattie Date: Wed, 24 Jun 2026 13:21:32 -0500 Subject: [PATCH 11/11] Include ready hook path diagnostics --- src/snapclass/schemas.py | 1 + tests/test_behavior_contracts.py | 1 + 2 files changed, 2 insertions(+) diff --git a/src/snapclass/schemas.py b/src/snapclass/schemas.py index 5dc488f..357d988 100644 --- a/src/snapclass/schemas.py +++ b/src/snapclass/schemas.py @@ -444,6 +444,7 @@ def _mark_snapshot_ready(instance: object) -> None: instance, "__snapclass_ready__", snapshot=snapshot, + path=expected_path, ) _ensure_snapshot_path_unchanged( snapshot, diff --git a/tests/test_behavior_contracts.py b/tests/test_behavior_contracts.py index 191b4bd..3267a12 100644 --- a/tests/test_behavior_contracts.py +++ b/tests/test_behavior_contracts.py @@ -396,6 +396,7 @@ def __snapclass_ready__(self, *, snapshot): ready_message = str(ready_error.value) assert "__snapclass_ready__" in ready_message + assert "ready.yml" in ready_message assert "ReadyItem" in ready_message assert "ready failed" in ready_message