Coverage for backend/django/flowsheetInternals/unitops/services/edit_operations/recorder.py: 88%
221 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from __future__ import annotations
3import json
4from dataclasses import dataclass
5from typing import Any, Iterable
7from django.core.serializers.json import DjangoJSONEncoder
8from django.db import models
9from django.db.models.signals import (
10 m2m_changed,
11 post_delete,
12 post_save,
13 pre_delete,
14 pre_save,
15)
17from .context import get_active_recorder
18from .scope import (
19 HistoryModelSpec,
20 affected_ids_for_instance,
21 get_history_model_spec,
22 history_model_specs,
23 resolve_instance_flowsheet_id,
24 resolve_instance_flowsheet_state_id,
25)
26from .types import OperationTransition
29def _json_value(value: Any) -> Any:
30 """Normalize Django/Python values to deterministic JSON-compatible data."""
31 return json.loads(json.dumps(value, cls=DjangoJSONEncoder))
34def snapshot_instance(
35 instance: models.Model,
36 spec: HistoryModelSpec | None = None,
37) -> dict[str, Any]:
38 """Serialize every replayable concrete field using database field names."""
39 spec = spec or get_history_model_spec(type(instance))
40 if spec is None: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true
41 raise ValueError(f"{instance._meta.label} is not tracked by flowsheet history.")
42 return {
43 field.attname: _json_value(getattr(instance, field.attname))
44 for field in spec.fields
45 }
48def _load_instance(model: type[models.Model], pk: Any) -> models.Model | None:
49 """Load rows through the unfiltered base manager, including tombstones."""
50 return model._base_manager.filter(pk=pk).first()
53@dataclass
54class RowChange:
55 """Original and final state for one row within a single user action."""
57 spec: HistoryModelSpec
58 pk: Any
59 before: dict[str, Any] | None
60 after: dict[str, Any] | None
61 before_object_ids: set[int]
62 before_group_ids: set[int]
63 after_object_ids: set[int]
64 after_group_ids: set[int]
67class FlowsheetChangeRecorder:
68 """Collect included row writes made inside one atomic flowsheet edit."""
70 def __init__(self, flowsheet_id: int, flowsheet_state_id: int, *, kind: str):
71 self.flowsheet_id = int(flowsheet_id)
72 self.flowsheet_state_id = int(flowsheet_state_id)
73 self.kind = kind
74 self._changes: dict[tuple[str, Any], RowChange] = {}
76 def _validate_owner(self, instance: models.Model) -> HistoryModelSpec | None:
77 spec = get_history_model_spec(type(instance))
78 if spec is None: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true
79 return None
80 owner_id = resolve_instance_flowsheet_id(instance)
81 state_id = resolve_instance_flowsheet_state_id(instance)
82 if owner_id != self.flowsheet_id or state_id != self.flowsheet_state_id: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 raise ValueError(
84 f"{spec.label} row {instance.pk!r} belongs to flowsheet/state "
85 f"{owner_id!r}/{state_id!r}, not active flowsheet/state "
86 f"{self.flowsheet_id}/{self.flowsheet_state_id}."
87 )
88 return spec
90 def record_before(
91 self,
92 instance: models.Model,
93 *,
94 row: dict[str, Any] | None = None,
95 ) -> None:
96 """Keep the first state observed for an existing row."""
97 spec = self._validate_owner(instance)
98 if spec is None or instance.pk is None: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true
99 return
100 key = (spec.label, _json_value(instance.pk))
101 if key not in self._changes:
102 before = snapshot_instance(instance, spec) if row is None else row
103 object_ids, group_ids = affected_ids_for_instance(instance)
104 self._changes[key] = RowChange(
105 spec,
106 key[1],
107 before,
108 before,
109 object_ids,
110 group_ids,
111 set(object_ids),
112 set(group_ids),
113 )
115 def record_created(self, instance: models.Model) -> None:
116 """Record a row whose identity was allocated inside this operation."""
117 spec = self._validate_owner(instance)
118 if spec is None or instance.pk is None: 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true
119 return
120 key = (spec.label, _json_value(instance.pk))
121 after = snapshot_instance(instance, spec)
122 object_ids, group_ids = affected_ids_for_instance(instance)
123 change = self._changes.get(key)
124 if change is None: 124 ↛ 136line 124 didn't jump to line 136 because the condition on line 124 was always true
125 self._changes[key] = RowChange(
126 spec,
127 key[1],
128 None,
129 after,
130 set(),
131 set(),
132 object_ids,
133 group_ids,
134 )
135 else:
136 change.after = after
137 change.after_object_ids = object_ids
138 change.after_group_ids = group_ids
140 def record_after(self, instance: models.Model) -> None:
141 """Store the latest state for a row changed one or more times."""
142 spec = self._validate_owner(instance)
143 if spec is None or instance.pk is None: 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true
144 return
145 key = (spec.label, _json_value(instance.pk))
146 after = snapshot_instance(instance, spec)
147 object_ids, group_ids = affected_ids_for_instance(instance)
148 change = self._changes.get(key)
149 if change is None: 149 ↛ 151line 149 didn't jump to line 151 because the condition on line 149 was never true
150 # Defensive fallback for a custom save path that skipped pre_save.
151 self._changes[key] = RowChange(
152 spec,
153 key[1],
154 after,
155 after,
156 object_ids,
157 group_ids,
158 set(object_ids),
159 set(group_ids),
160 )
161 else:
162 change.after = after
163 change.after_object_ids = object_ids
164 change.after_group_ids = group_ids
166 def record_deleted(self, instance: models.Model) -> None:
167 """Set the final state to absence while preserving the first before image."""
168 spec = self._validate_owner(instance)
169 if spec is None or instance.pk is None: 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true
170 return
171 key = (spec.label, _json_value(instance.pk))
172 if key not in self._changes: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true
173 before = snapshot_instance(instance, spec)
174 object_ids, group_ids = affected_ids_for_instance(instance)
175 self._changes[key] = RowChange(
176 spec,
177 key[1],
178 before,
179 None,
180 object_ids,
181 group_ids,
182 set(),
183 set(),
184 )
185 else:
186 change = self._changes[key]
187 change.after = None
188 change.after_object_ids = set()
189 change.after_group_ids = set()
191 def _payload_change(
192 self,
193 change: RowChange,
194 *,
195 forward: bool,
196 ) -> dict[str, Any] | None:
197 # A product-level GROUP keeps its Grouping identity behind the group's
198 # SimulationObject tombstone. This preserves the established stable-ID
199 # contract while every other created grouping (including custom zones)
200 # follows ordinary create/delete replay.
201 if (
202 self.kind == "group"
203 and change.spec.label == "flowsheetinternals_graphicdata.grouping"
204 and change.before is None
205 ):
206 return None
207 source = change.before if forward else change.after
208 target = change.after if forward else change.before
209 if source == target:
210 return None
211 if source is None:
212 action = "create"
213 target_fields = target
214 elif target is None:
215 action = "delete"
216 target_fields = None
217 else:
218 action = "update"
219 target_fields = {
220 field: target[field]
221 for field in target
222 if source.get(field) != target.get(field)
223 }
224 if not target_fields: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true
225 return None
226 return {
227 "model": change.spec.label,
228 "pk": change.pk,
229 "action": action,
230 "fields": target_fields,
231 }
233 def build_transition(self) -> OperationTransition | None:
234 """Return one minimal generic transition, or ``None`` for a no-op."""
235 ordered_changes = sorted(
236 self._changes.values(),
237 key=lambda change: (change.spec.label, str(change.pk)),
238 )
239 material_changes = [
240 (change, payload)
241 for change in ordered_changes
242 if (payload := self._payload_change(change, forward=True)) is not None
243 ]
244 forward = [payload for _, payload in material_changes]
245 if not forward:
246 return None
247 inverse = [
248 payload
249 for change, _ in material_changes
250 if (payload := self._payload_change(change, forward=False)) is not None
251 ]
252 affected_object_ids = {
253 object_id
254 for change, _ in material_changes
255 for object_id in change.before_object_ids | change.after_object_ids
256 }
257 affected_group_ids = {
258 group_id
259 for change, _ in material_changes
260 for group_id in change.before_group_ids | change.after_group_ids
261 }
262 if self.kind == "duplicate":
263 # Duplicate responses define their affected batch as the copied
264 # SimulationObject identities. References intentionally retained
265 # to unselected source streams are not part of that copy batch.
266 affected_object_ids = {
267 int(change.pk)
268 for change, _ in material_changes
269 if change.spec.label == "flowsheetinternals_unitops.simulationobject"
270 and change.before is None
271 }
272 return OperationTransition(
273 forward_payload={"handler": "row_change_set", "changes": forward},
274 inverse_payload={"handler": "row_change_set", "changes": inverse},
275 affected_object_ids=affected_object_ids,
276 affected_group_ids=affected_group_ids,
277 )
280def _active_for(instance: models.Model) -> FlowsheetChangeRecorder | None:
281 recorder = get_active_recorder()
282 if recorder is None or get_history_model_spec(type(instance)) is None:
283 return None
284 return recorder
287def _history_pre_save(sender, instance, raw=False, **kwargs) -> None:
288 recorder = _active_for(instance)
289 if recorder is None or raw or instance._state.adding or instance.pk is None:
290 return
291 existing = _load_instance(sender, instance.pk)
292 if existing is not None: 292 ↛ exitline 292 didn't return from function '_history_pre_save' because the condition on line 292 was always true
293 recorder.record_before(existing)
296def _history_post_save(sender, instance, created=False, raw=False, **kwargs) -> None:
297 recorder = _active_for(instance)
298 if recorder is None or raw:
299 return
300 persisted = _load_instance(sender, instance.pk)
301 if persisted is None: 301 ↛ 302line 301 didn't jump to line 302 because the condition on line 301 was never true
302 raise RuntimeError(
303 f"Could not load saved {sender._meta.label} row {instance.pk!r}."
304 )
305 if created:
306 recorder.record_created(persisted)
307 else:
308 # The in-memory instance may contain values omitted from update_fields,
309 # or unresolved database expressions. Journal only committed row state.
310 recorder.record_after(persisted)
313def _history_pre_delete(sender, instance, **kwargs) -> None:
314 recorder = _active_for(instance)
315 if recorder is not None:
316 # Django may fast-delete automatic M2M rows during a parent cascade
317 # without emitting m2m_changed or per-row delete signals. Snapshot
318 # those links while both endpoints still exist.
319 for through_spec in history_model_specs().values():
320 if not through_spec.is_automatic_through_model:
321 continue
322 owner_fields = [
323 field
324 for field in through_spec.fields
325 if isinstance(field, models.ForeignKey)
326 and field.remote_field.model is sender
327 ]
328 for owner_field in owner_fields:
329 for through_row in through_spec.model._base_manager.filter(
330 **{owner_field.attname: instance.pk}
331 ):
332 recorder.record_before(through_row)
333 recorder.record_deleted(through_row)
334 recorder.record_before(instance)
337def _history_post_delete(sender, instance, **kwargs) -> None:
338 recorder = _active_for(instance)
339 if recorder is not None:
340 recorder.record_deleted(instance)
343def _through_rows(sender, instance, reverse: bool, pk_set) -> list[models.Model]:
344 relation_fields = [
345 field
346 for field in sender._meta.concrete_fields
347 if isinstance(field, models.ForeignKey)
348 ]
349 source_field = next(
350 field for field in relation_fields if field.remote_field.model is type(instance)
351 )
352 target_field = next(field for field in relation_fields if field is not source_field)
353 filters = {source_field.attname: instance.pk}
354 if pk_set is not None: 354 ↛ 356line 354 didn't jump to line 356 because the condition on line 354 was always true
355 filters[f"{target_field.attname}__in"] = pk_set
356 return list(sender._base_manager.filter(**filters))
359def _history_m2m_changed(
360 sender,
361 instance,
362 action,
363 reverse,
364 model,
365 pk_set,
366 **kwargs,
367) -> None:
368 recorder = get_active_recorder()
369 if recorder is None or get_history_model_spec(sender) is None:
370 return
371 if action in {"pre_remove", "pre_clear"}: 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true
372 for row in _through_rows(sender, instance, reverse, pk_set):
373 recorder.record_before(row)
374 recorder.record_deleted(row)
375 elif action == "post_add":
376 for row in _through_rows(sender, instance, reverse, pk_set):
377 recorder.record_created(row)
380def connect_history_signals() -> None:
381 """Install global receivers; model scope filtering keeps them inexpensive."""
382 pre_save.connect(_history_pre_save, dispatch_uid="flowsheet_history_pre_save")
383 post_save.connect(_history_post_save, dispatch_uid="flowsheet_history_post_save")
384 pre_delete.connect(
385 _history_pre_delete,
386 dispatch_uid="flowsheet_history_pre_delete",
387 )
388 post_delete.connect(
389 _history_post_delete,
390 dispatch_uid="flowsheet_history_post_delete",
391 )
392 m2m_changed.connect(
393 _history_m2m_changed,
394 dispatch_uid="flowsheet_history_m2m_changed",
395 )
398def tracked_queryset_update(queryset, **fields) -> int:
399 """Run `QuerySet.update` while recording before/after images when active."""
400 recorder = get_active_recorder()
401 spec = get_history_model_spec(queryset.model)
402 if recorder is None or spec is None:
403 return queryset.update(**fields)
404 before_instances = list(queryset)
405 for instance in before_instances:
406 recorder.record_before(instance)
407 pks = [instance.pk for instance in before_instances]
408 updated = queryset.update(**fields)
409 for instance in queryset.model._base_manager.filter(pk__in=pks):
410 recorder.record_after(instance)
411 return updated
414def tracked_bulk_create(
415 manager,
416 objects: Iterable[models.Model],
417 **kwargs,
418) -> list[models.Model]:
419 """Run `bulk_create` and journal every allocated included row."""
420 object_list = list(objects)
421 created = manager.bulk_create(object_list, **kwargs)
422 recorder = get_active_recorder()
423 if recorder is not None and get_history_model_spec(manager.model) is not None:
424 if any(instance.pk is None for instance in created): 424 ↛ 425line 424 didn't jump to line 425 because the condition on line 424 was never true
425 raise RuntimeError(
426 "Tracked bulk_create must return primary keys for every row."
427 )
428 for instance in created:
429 recorder.record_created(instance)
430 return created
433def tracked_bulk_update(
434 manager,
435 objects: Iterable[models.Model],
436 fields: Iterable[str],
437 **kwargs,
438) -> int:
439 """Run `bulk_update` and journal all changed included rows."""
440 object_list = list(objects)
441 recorder = get_active_recorder()
442 spec = get_history_model_spec(manager.model)
443 pks = [instance.pk for instance in object_list]
444 if recorder is not None and spec is not None:
445 for instance in manager.model._base_manager.filter(pk__in=pks):
446 recorder.record_before(instance)
447 updated = manager.bulk_update(object_list, list(fields), **kwargs)
448 if recorder is not None and spec is not None:
449 for instance in manager.model._base_manager.filter(pk__in=pks):
450 recorder.record_after(instance)
451 return updated