Coverage for backend/django/flowsheetInternals/unitops/services/edit_operations/replay.py: 82%
213 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from __future__ import annotations
3from collections.abc import Iterable
4from typing import Any
6from django.db import models
7from rest_framework.exceptions import ValidationError
9from core.auxiliary.models.Flowsheet import Flowsheet
11from .context import replay_in_progress
12from .lifecycle import after_replay, before_replay_deletions
13from .scope import (
14 HistoryModelSpec,
15 get_history_model_spec,
16 resolve_instance_flowsheet_id,
17 resolve_instance_flowsheet_state_id,
18 resolve_row_flowsheet_id,
19 resolve_row_flowsheet_state_id,
20)
23VALID_ACTIONS = frozenset({"create", "update", "delete"})
26def _python_value(field: models.Field, value: Any) -> Any:
27 if value is None:
28 return None
29 return field.to_python(value)
32def _deserialize_fields(
33 spec: HistoryModelSpec,
34 values: dict[str, Any],
35) -> dict[str, Any]:
36 fields_by_name = {field.attname: field for field in spec.fields}
37 unknown = set(values) - set(fields_by_name)
38 if unknown: 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true
39 raise ValidationError(
40 {"detail": f"This edit contains unsupported fields: {sorted(unknown)}."}
41 )
42 return {
43 name: _python_value(fields_by_name[name], value)
44 for name, value in values.items()
45 }
48def _validate_create_identity(
49 spec: HistoryModelSpec,
50 *,
51 pk: Any,
52 fields: dict[str, Any],
53) -> None:
54 """Require the payload identity to match the complete created row."""
55 pk_name = spec.model._meta.pk.attname
56 if str(fields[pk_name]) != str(pk): 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true
57 raise ValidationError(
58 {"detail": "A created history row has an inconsistent identity."}
59 )
62def _normalize_changes(payload: dict, flowsheet: Flowsheet) -> list[dict]:
63 if payload.get("handler") != "row_change_set": 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true
64 raise ValidationError({"detail": "This edit operation type is not supported."})
65 raw_changes = payload.get("changes")
66 if not isinstance(raw_changes, list): 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true
67 raise ValidationError({"detail": "The edit operation payload is invalid."})
69 normalized: list[dict] = []
70 seen: set[tuple[str, str]] = set()
71 for raw_change in raw_changes:
72 if not isinstance(raw_change, dict): 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true
73 raise ValidationError({"detail": "The edit operation payload is invalid."})
74 model_label = str(raw_change.get("model", "")).lower()
75 spec = get_history_model_spec(model_label)
76 action = raw_change.get("action")
77 pk = raw_change.get("pk")
78 if spec is None or action not in VALID_ACTIONS or pk is None: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true
79 raise ValidationError({"detail": "The edit operation payload is invalid."})
80 key = (model_label, str(pk))
81 if key in seen: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 raise ValidationError({"detail": "The edit operation payload is invalid."})
83 seen.add(key)
85 fields = raw_change.get("fields")
86 if action == "delete":
87 if fields is not None: 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true
88 raise ValidationError(
89 {"detail": "The edit operation payload is invalid."}
90 )
91 deserialized_fields = None
92 else:
93 if not isinstance(fields, dict): 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 raise ValidationError(
95 {"detail": "The edit operation payload is invalid."}
96 )
97 deserialized_fields = _deserialize_fields(spec, fields)
98 if action == "create" and set(deserialized_fields) != spec.field_names: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true
99 raise ValidationError(
100 {"detail": "A created history row does not contain complete state."}
101 )
102 if action == "create":
103 _validate_create_identity(
104 spec,
105 pk=pk,
106 fields=deserialized_fields,
107 )
108 if ( 108 ↛ 112line 108 didn't jump to line 112 because the condition on line 108 was never true
109 action == "update"
110 and spec.model._meta.pk.attname in deserialized_fields
111 ):
112 raise ValidationError(
113 {"detail": "An edit cannot change a row identity."}
114 )
116 existing = spec.model._base_manager.filter(pk=pk).first()
117 if action != "create":
118 if existing is None:
119 raise ValidationError(
120 {"detail": "A row referenced by this edit no longer exists."}
121 )
122 normalized.append(
123 {
124 "model": model_label,
125 "pk": pk,
126 "action": action,
127 "fields": deserialized_fields,
128 "spec": spec,
129 "existing": existing,
130 }
131 )
133 pending_creates = {
134 (change["model"], str(change["pk"])): change
135 for change in normalized
136 if change["action"] == "create"
137 }
139 def pending_owner(change: dict, visiting: set[tuple[str, str]]) -> int | None:
140 key = (change["model"], str(change["pk"]))
141 if key in visiting: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 return None
143 visiting.add(key)
144 fields = change["fields"] or {}
145 owner_id = resolve_row_flowsheet_id(change["spec"], fields)
146 if owner_id is not None:
147 visiting.remove(key)
148 return owner_id
149 for field in change["spec"].fields: 149 ↛ 162line 149 didn't jump to line 162 because the loop on line 149 didn't complete
150 if not isinstance(field, models.ForeignKey):
151 continue
152 related_id = fields.get(field.attname)
153 related_spec = get_history_model_spec(field.remote_field.model)
154 if related_id is None or related_spec is None: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 continue
156 related_change = pending_creates.get((related_spec.label, str(related_id)))
157 if related_change is not None: 157 ↛ 149line 157 didn't jump to line 149 because the condition on line 157 was always true
158 owner_id = pending_owner(related_change, visiting)
159 if owner_id is not None: 159 ↛ 149line 159 didn't jump to line 149 because the condition on line 159 was always true
160 visiting.remove(key)
161 return owner_id
162 visiting.remove(key)
163 return None
165 def pending_state(change: dict, visiting: set[tuple[str, str]]) -> int | None:
166 """Resolve state ownership through rows created in this payload."""
167 key = (change["model"], str(change["pk"]))
168 if key in visiting: 168 ↛ 169line 168 didn't jump to line 169 because the condition on line 168 was never true
169 return None
170 visiting.add(key)
171 fields = change["fields"] or {}
172 state_id = resolve_row_flowsheet_state_id(change["spec"], fields)
173 if state_id is not None:
174 visiting.remove(key)
175 return state_id
176 for field in change["spec"].fields: 176 ↛ 189line 176 didn't jump to line 189 because the loop on line 176 didn't complete
177 if not isinstance(field, models.ForeignKey):
178 continue
179 related_id = fields.get(field.attname)
180 related_spec = get_history_model_spec(field.remote_field.model)
181 if related_id is None or related_spec is None: 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true
182 continue
183 related_change = pending_creates.get((related_spec.label, str(related_id)))
184 if related_change is not None: 184 ↛ 176line 184 didn't jump to line 176 because the condition on line 184 was always true
185 state_id = pending_state(related_change, visiting)
186 if state_id is not None: 186 ↛ 176line 186 didn't jump to line 176 because the condition on line 186 was always true
187 visiting.remove(key)
188 return state_id
189 visiting.remove(key)
190 return None
192 def validate_foreign_keys(change: dict) -> None:
193 """Reject missing and cross-flowsheet relations before replay writes."""
194 fields = change["fields"] or {}
195 for field in change["spec"].fields:
196 if not isinstance(field, models.ForeignKey) or field.attname not in fields:
197 continue
198 related_id = fields[field.attname]
199 if related_id is None:
200 continue
202 related_model = field.remote_field.model
203 related_spec = get_history_model_spec(related_model)
204 related_change = (
205 pending_creates.get((related_spec.label, str(related_id)))
206 if related_spec is not None
207 else None
208 )
209 if related_change is not None:
210 related_owner_id = pending_owner(related_change, set())
211 related_state_id = pending_state(related_change, set())
212 else:
213 related_instance = related_model._base_manager.filter(
214 pk=related_id
215 ).first()
216 if related_instance is None:
217 raise ValidationError(
218 {"detail": "This edit references a row that no longer exists."}
219 )
220 if related_model is Flowsheet: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true
221 related_owner_id = related_instance.pk
222 related_state_id = None
223 elif related_spec is not None:
224 related_owner_id = resolve_instance_flowsheet_id(related_instance)
225 related_state_id = resolve_instance_flowsheet_state_id(
226 related_instance
227 )
228 else:
229 related_owner_id = getattr(
230 related_instance,
231 "flowsheet_id",
232 None,
233 )
234 related_state_id = (
235 related_instance.pk
236 if related_model._meta.label_lower
237 == "core_auxiliary.flowsheetstate"
238 else getattr(related_instance, "flowsheet_state_id", None)
239 )
241 if related_owner_id is not None and related_owner_id != flowsheet.pk:
242 raise ValidationError(
243 {"detail": "This edit references data from another flowsheet."}
244 )
245 if ( 245 ↛ 249line 245 didn't jump to line 249 because the condition on line 245 was never true
246 related_state_id is not None
247 and related_state_id != flowsheet.current_state_id
248 ):
249 raise ValidationError(
250 {
251 "detail": "This edit references data from another flowsheet state."
252 }
253 )
255 for change in normalized:
256 validate_foreign_keys(change)
257 if change["action"] == "create":
258 if change["existing"] is not None:
259 soft_delete_field = change["spec"].soft_delete_field
260 if not soft_delete_field or not getattr( 260 ↛ 263line 260 didn't jump to line 263 because the condition on line 260 was never true
261 change["existing"], soft_delete_field
262 ):
263 raise ValidationError(
264 {"detail": "A row restored by this edit already exists."}
265 )
266 owner_id = resolve_instance_flowsheet_id(change["existing"])
267 state_id = resolve_instance_flowsheet_state_id(change["existing"])
268 else:
269 owner_id = pending_owner(change, set())
270 state_id = pending_state(change, set())
271 else:
272 owner_id = resolve_instance_flowsheet_id(change["existing"])
273 state_id = resolve_instance_flowsheet_state_id(change["existing"])
274 if owner_id != flowsheet.pk: 274 ↛ 275line 274 didn't jump to line 275 because the condition on line 274 was never true
275 raise ValidationError(
276 {"detail": "This edit references data from another flowsheet."}
277 )
278 if state_id != flowsheet.current_state_id: 278 ↛ 279line 278 didn't jump to line 279 because the condition on line 278 was never true
279 raise ValidationError(
280 {"detail": "This edit references data from another flowsheet state."}
281 )
282 change.pop("existing", None)
283 return normalized
286def _dependency_ranks(specs: Iterable[HistoryModelSpec]) -> dict[str, int]:
287 by_label = {spec.label: spec for spec in specs}
288 ranks: dict[str, int] = {}
290 def rank(label: str, visiting: set[str]) -> int:
291 if label in ranks:
292 return ranks[label]
293 if label in visiting: 293 ↛ 295line 293 didn't jump to line 295 because the condition on line 293 was never true
294 # Nullable/deferred cycles are applied in a later update phase.
295 return 0
296 visiting.add(label)
297 dependency_ranks = [
298 rank(dependency, visiting)
299 for dependency in by_label[label].dependencies
300 if dependency in by_label and dependency != label
301 ]
302 visiting.remove(label)
303 ranks[label] = (max(dependency_ranks) + 1) if dependency_ranks else 0
304 return ranks[label]
306 for label in by_label:
307 rank(label, set())
308 return ranks
311def _apply_delete(change: dict) -> None:
312 instance = change["spec"].model._base_manager.filter(pk=change["pk"]).first()
313 if instance is None: 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true
314 raise ValidationError(
315 {"detail": "A row referenced by this edit no longer exists."}
316 )
317 soft_delete_field = change["spec"].soft_delete_field
318 if soft_delete_field:
319 change["spec"].model._base_manager.filter(pk=change["pk"]).update(
320 **{soft_delete_field: True}
321 )
322 return
323 instance.delete()
326def _apply_create(change: dict) -> None:
327 existing = change["spec"].model._base_manager.filter(pk=change["pk"]).first()
328 if existing is None:
329 change["spec"].model(**change["fields"]).save(force_insert=True)
330 elif change["spec"].soft_delete_field: 330 ↛ 335line 330 didn't jump to line 335 because the condition on line 330 was always true
331 change["spec"].model._base_manager.filter(pk=change["pk"]).update(
332 **change["fields"]
333 )
334 else:
335 raise ValidationError({"detail": "A row restored by this edit already exists."})
338def _apply_update(change: dict) -> None:
339 updated = (
340 change["spec"]
341 .model._base_manager.filter(pk=change["pk"])
342 .update(**change["fields"])
343 )
344 if updated != 1: 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true
345 raise ValidationError(
346 {"detail": "A row referenced by this edit no longer exists."}
347 )
350def apply_operation_payload(payload: dict, *, flowsheet: Flowsheet) -> None:
351 """Validate and apply one generic row change-set atomically."""
352 changes = _normalize_changes(payload, flowsheet)
353 ranks = _dependency_ranks(change["spec"] for change in changes)
354 token = replay_in_progress.set(True)
355 try:
356 deleted_properties = before_replay_deletions(changes)
357 for change in sorted(
358 (change for change in changes if change["action"] == "delete"),
359 key=lambda change: ranks[change["model"]],
360 reverse=True,
361 ):
362 _apply_delete(change)
363 for change in sorted(
364 (change for change in changes if change["action"] == "create"),
365 key=lambda change: ranks[change["model"]],
366 ):
367 _apply_create(change)
368 for change in sorted(
369 (change for change in changes if change["action"] == "update"),
370 key=lambda change: ranks[change["model"]],
371 ):
372 _apply_update(change)
373 after_replay(changes, deleted_properties)
374 finally:
375 replay_in_progress.reset(token)