Coverage for src/objectstate/transaction_checkpoint.py: 99%
86 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 07:22 +0000
1"""Exact local rollback boundary for an ObjectState transaction participant."""
3from __future__ import annotations
5import copy
6from dataclasses import dataclass
7from typing import Any, TYPE_CHECKING
9from objectstate.construction_binding import StateConstructionBinding
10from objectstate.object_state_metadata import ObjectStateMetadataStore
11from objectstate.parameter_structure import ParameterStructure
12from objectstate.value_semantics import semantic_values_equal
14if TYPE_CHECKING:
15 from objectstate.object_state import ObjectState
18@dataclass(frozen=True, slots=True)
19class ObjectStateTransactionCheckpoint:
20 """Capture and restore one ObjectState without changing its edit semantics.
22 This is intentionally distinct from the serializable time-travel snapshot:
23 a local transaction checkpoint retains the exact saved object and delegate
24 identities as well as live unsaved edits and ephemeral reconstruction
25 caches.
26 """
28 object_instance: Any
29 construction: StateConstructionBinding
30 parent_state: ObjectState | None
31 structure: ParameterStructure
32 preserve_value_identity: bool
33 parameters: dict[str, Any]
34 saved_parameters: dict[str, Any]
35 live_resolved: dict[str, Any] | None
36 saved_resolved: dict[str, Any]
37 live_provenance: dict[str, Any]
38 metadata: dict[str, Any]
39 cached_object: Any
40 cached_object_applied: bool
41 invalid_fields: frozenset[str]
42 resolved_from: dict[bool, Any]
43 raw_dirty: bool
44 dirty_fields: frozenset[str]
45 signature_diff_fields: frozenset[str]
46 last_changed_field: str | None
47 last_changed_concrete_paths: frozenset[str]
48 last_changed_paths: frozenset[str]
49 last_changed_values: dict[str, Any]
50 last_changed_meta_keys: frozenset[str]
51 last_changed_meta_values: dict[str, Any]
53 @classmethod
54 def capture(
55 cls, state: ObjectState, *, preserve_value_identity: bool = False
56 ) -> "ObjectStateTransactionCheckpoint":
57 """Capture the exact local state before a fallible transaction."""
59 copied = cls._checkpoint_mappings(
60 state,
61 {
62 "parameters": state.parameters,
63 "saved_parameters": state._saved_parameters,
64 "live_resolved": state._live_resolved,
65 "saved_resolved": state._saved_resolved,
66 "live_provenance": state._live_provenance,
67 "last_changed_values": state._last_changed_values,
68 "last_changed_meta_values": state._last_changed_meta_values,
69 },
70 preserve_value_identity=preserve_value_identity,
71 )
72 return cls(
73 object_instance=state.object_instance,
74 construction=state.construction_binding(),
75 parent_state=state._parent_state,
76 structure=state.parameter_structure.clone(),
77 preserve_value_identity=preserve_value_identity,
78 parameters=copied["parameters"],
79 saved_parameters=copied["saved_parameters"],
80 live_resolved=copied["live_resolved"],
81 saved_resolved=copied["saved_resolved"],
82 live_provenance=copied["live_provenance"],
83 metadata=state.copy_metadata_for_snapshot(),
84 cached_object=state._cached_object,
85 cached_object_applied=state._cached_object_applied,
86 invalid_fields=frozenset(state._invalid_fields),
87 resolved_from=dict(state._resolved_from),
88 raw_dirty=state._raw_dirty,
89 dirty_fields=frozenset(state._dirty_fields),
90 signature_diff_fields=frozenset(state._signature_diff_fields),
91 last_changed_field=state._last_changed_field,
92 last_changed_concrete_paths=frozenset(state._last_changed_concrete_paths),
93 last_changed_paths=frozenset(state._last_changed_paths),
94 last_changed_values=copied["last_changed_values"],
95 last_changed_meta_keys=frozenset(state._last_changed_meta_keys),
96 last_changed_meta_values=copied["last_changed_meta_values"],
97 )
99 @property
100 def extraction_target(self) -> Any:
101 return self.construction.target
103 def restore(self, state: ObjectState, *, emit_notifications: bool = True) -> None:
104 """Restore the captured state and then publish one coherent change."""
106 changed_fields = self._changed_fields(state) if emit_notifications else set()
107 copied = self._checkpoint_mappings(
108 state,
109 {
110 "parameters": self.parameters,
111 "saved_parameters": self.saved_parameters,
112 "live_resolved": self.live_resolved,
113 "saved_resolved": self.saved_resolved,
114 "live_provenance": self.live_provenance,
115 "last_changed_values": self.last_changed_values,
116 "last_changed_meta_values": self.last_changed_meta_values,
117 },
118 preserve_value_identity=self.preserve_value_identity,
119 )
121 state.object_instance = self.object_instance
122 self.construction.attach(state)
123 state._parent_state = self.parent_state
124 state._parent_field_name = self.construction.parent_field
125 state.parameter_structure = self.structure.clone()
126 state.parameters = copied["parameters"]
127 state._index_parameter_paths()
128 state._saved_parameters = copied["saved_parameters"]
129 state._live_resolved = copied["live_resolved"]
130 state._saved_resolved = copied["saved_resolved"]
131 state._live_provenance = copied["live_provenance"]
132 state.metadata = ObjectStateMetadataStore.from_snapshot(
133 scope_id=state.scope_id,
134 metadata=self.metadata,
135 )
136 state._cached_object = self.cached_object
137 state._cached_object_applied = self.cached_object_applied
138 state._invalid_fields = set(self.invalid_fields)
139 state._resolved_from = dict(self.resolved_from)
140 state._raw_dirty = self.raw_dirty
141 state._dirty_fields = set(self.dirty_fields)
142 state._signature_diff_fields = set(self.signature_diff_fields)
143 state._last_changed_field = self.last_changed_field
144 state._last_changed_concrete_paths = set(self.last_changed_concrete_paths)
145 state._last_changed_paths = set(self.last_changed_paths)
146 state._last_changed_values = copied["last_changed_values"]
147 state._last_changed_meta_keys = set(self.last_changed_meta_keys)
148 state._last_changed_meta_values = copied["last_changed_meta_values"]
150 if changed_fields:
151 state._notify_resolved_changed(
152 changed_fields,
153 context="transaction_checkpoint_restore",
154 )
155 state._notify_state_changed(changed_fields)
157 def _changed_fields(self, state: ObjectState) -> set[str]:
158 fields: set[str] = set()
159 missing = object()
160 for current, target in (
161 (state.parameters, self.parameters),
162 (state._saved_parameters, self.saved_parameters),
163 (state._live_resolved or {}, self.live_resolved or {}),
164 (state._saved_resolved, self.saved_resolved),
165 ):
166 fields.update(
167 key
168 for key in set(current) | set(target)
169 if not semantic_values_equal(
170 current.get(key, missing),
171 target.get(key, missing),
172 )
173 )
174 return state._most_specific_notification_fields(fields)
176 @staticmethod
177 def _checkpoint_mappings(
178 state: ObjectState,
179 mappings: dict[str, Any],
180 *,
181 preserve_value_identity: bool,
182 ) -> dict[str, Any]:
183 if preserve_value_identity:
184 return mappings
185 memo: dict[int, Any] = {}
186 seen: set[int] = set()
187 state._seed_callable_identity_memo(mappings, memo, seen)
188 return copy.deepcopy(mappings, memo)