Skip to content
38 changes: 38 additions & 0 deletions skills/snapclass-fluency/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 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:

```python
Expand Down Expand Up @@ -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:

Expand Down Expand Up @@ -1096,6 +1130,10 @@ 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.
- 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.
Expand Down
20 changes: 15 additions & 5 deletions src/snapclass/collections.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,20 +43,29 @@ 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)
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)
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]:
from .schemas import _PatternMatcher, _attach_snapshot, _has_path_value
from .schemas import (
_PatternMatcher,
_attach_snapshot,
_has_path_value,
)

__tracebackhide__ = sessions.HIDDEN_TRACEBACK
if not self.model.__snapclass_config__.pattern:
Expand All @@ -76,7 +85,8 @@ 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
instance.snapshot.load(_initial=True)
yield instance
else:
yield self.get(*values)
Expand Down
2 changes: 2 additions & 0 deletions src/snapclass/hooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
161 changes: 145 additions & 16 deletions src/snapclass/schemas.py
Original file line number Diff line number Diff line change
Expand Up @@ -355,7 +355,8 @@ def sync(
if not hasattr(cls, "__snapclass_config__"):
_install(cls, config)
_attach_snapshot(instance, config)
if _auto_enabled(config):
_mark_snapshot_ready(instance)
if _auto_enabled(config, instance):
instance.snapshot.save()
return instance

Expand Down Expand Up @@ -428,6 +429,117 @@ 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
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=expected_path,
)
Comment thread
Mattie marked this conversation as resolved.
_ensure_snapshot_path_unchanged(
snapshot,
expected_path,
"__snapclass_ready__",
)
except Exception:
snapshot._ready = False
raise


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
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(
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):
Comment thread
Mattie marked this conversation as resolved.
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 saves and reloads for one instance during lifecycle hooks."""
previous_loading = getattr(instance, "_snapclass_loading", 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)


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:
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:
cls.__snapclass_config__ = config
cls.snapshots = CollectionDescriptor()
Expand All @@ -449,11 +561,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):
if self.snapshot.exists:
self.snapshot.load(_initial=True)
else:
self.snapshot.save()
automatic = _auto_enabled(config, self)
if automatic and self.snapshot.exists:
self.snapshot.load(_initial=True)
else:
_mark_snapshot_ready(self)
if automatic:
if self.snapshot.exists:
Comment thread
Mattie marked this conversation as resolved.
self.snapshot.load()
else:
self.snapshot.save()

cls.__init__ = __init__

Expand All @@ -469,6 +586,11 @@ 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):
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)
Expand All @@ -480,7 +602,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
Expand Down Expand Up @@ -513,7 +635,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()

Expand All @@ -531,7 +653,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
Expand Down Expand Up @@ -605,6 +727,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:
Expand Down Expand Up @@ -794,6 +917,8 @@ 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)
Comment thread
Mattie marked this conversation as resolved.
finally:
object.__setattr__(self._instance, "_snapclass_loading", False)

Expand Down Expand Up @@ -834,7 +959,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)

Expand Down Expand Up @@ -895,7 +1020,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)

Expand Down Expand Up @@ -965,7 +1090,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)

Expand Down Expand Up @@ -1030,7 +1155,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)

Expand Down Expand Up @@ -1116,7 +1241,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)

Expand Down Expand Up @@ -2584,8 +2709,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:
Expand Down
Loading