Coverage for backend/django/flowsheetInternals/unitops/services/edit_operations/scope.py: 90%
188 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from __future__ import annotations
3from dataclasses import dataclass
4from functools import lru_cache
5from typing import Any
7from django.apps import apps
8from django.db import models
10from core.auxiliary.models.FlowsheetHistoryModel import (
11 DependentFlowsheetHistoryModel,
12 FlowsheetHistoryModel,
13)
16PRIMARY = "primary"
17DEPENDENT = "dependent"
20@dataclass(frozen=True)
21class HistoryModelSpec:
22 """Replay and ownership metadata for one journal-managed table."""
24 model: type[models.Model]
25 scope: str
26 fields: tuple[models.Field, ...]
27 dependencies: tuple[str, ...]
28 is_automatic_through_model: bool = False
29 soft_delete_field: str | None = None
31 @property
32 def label(self) -> str:
33 return self.model._meta.label_lower.lower()
35 @property
36 def field_names(self) -> frozenset[str]:
37 return frozenset(field.attname for field in self.fields)
40def _is_primary(model: type[models.Model]) -> bool:
41 return issubclass(model, FlowsheetHistoryModel)
44def _is_dependent(model: type[models.Model]) -> bool:
45 return issubclass(model, DependentFlowsheetHistoryModel)
48def _tracked_scope(model: type[models.Model]) -> str | None:
49 if _is_primary(model):
50 return PRIMARY
51 if _is_dependent(model):
52 return DEPENDENT
53 return None
56def _is_tracked_through_model(model: type[models.Model]) -> bool:
57 owner = model._meta.auto_created
58 if not isinstance(owner, type) or not _is_primary(owner):
59 return False
60 related_models = {
61 field.remote_field.model
62 for field in model._meta.concrete_fields
63 if isinstance(field, models.ForeignKey)
64 }
65 return bool(related_models) and all(
66 _is_primary(related) for related in related_models
67 )
70def _replayable_fields(model: type[models.Model]) -> tuple[models.Field, ...]:
71 """Return concrete state fields stored in generic history payloads."""
72 return tuple(
73 field
74 for field in model._meta.concrete_fields
75 if not getattr(field, "auto_now", False)
76 and not getattr(field, "auto_now_add", False)
77 )
80@lru_cache(maxsize=1)
81def history_model_specs() -> dict[str, HistoryModelSpec]:
82 """Discover primary, dependent, and automatic through-table participants."""
83 tracked_models: list[tuple[type[models.Model], str, bool]] = []
84 for model in apps.get_models(include_auto_created=True):
85 scope = _tracked_scope(model)
86 is_through = False
87 if scope is None and _is_tracked_through_model(model):
88 scope = PRIMARY
89 is_through = True
90 if scope is not None:
91 tracked_models.append((model, scope, is_through))
93 tracked_labels = {model._meta.label_lower.lower() for model, _, _ in tracked_models}
94 specs: dict[str, HistoryModelSpec] = {}
95 for model, scope, is_through in tracked_models:
96 dependencies = tuple(
97 sorted(
98 {
99 field.remote_field.model._meta.label_lower.lower()
100 for field in model._meta.concrete_fields
101 if isinstance(field, models.ForeignKey)
102 and field.remote_field.model._meta.label_lower.lower()
103 in tracked_labels
104 }
105 )
106 )
107 spec = HistoryModelSpec(
108 model=model,
109 scope=scope,
110 fields=_replayable_fields(model),
111 dependencies=dependencies,
112 is_automatic_through_model=is_through,
113 soft_delete_field=getattr(
114 model, "flowsheet_history_soft_delete_field", None
115 ),
116 )
117 specs[spec.label] = spec
118 return specs
121def get_history_model_spec(
122 model_or_label: type[models.Model] | str,
123) -> HistoryModelSpec | None:
124 """Return history metadata for a model, or ``None`` when it is excluded."""
125 label = (
126 model_or_label
127 if isinstance(model_or_label, str)
128 else model_or_label._meta.label_lower
129 )
130 return history_model_specs().get(label.lower())
133def _base_queryset(model: type[models.Model]):
134 """Use the unfiltered base manager so soft-deleted rows remain replayable."""
135 return model._base_manager.all()
138def resolve_row_flowsheet_id(
139 spec: HistoryModelSpec,
140 row: dict[str, Any],
141) -> int | None:
142 """Resolve the owning flowsheet from a normalized row snapshot."""
143 state_id = resolve_row_flowsheet_state_id(spec, row)
144 if state_id is not None:
145 state_model = apps.get_model("core_auxiliary", "FlowsheetState")
146 value = (
147 state_model._base_manager.filter(pk=state_id)
148 .values_list("flowsheet_id", flat=True)
149 .first()
150 )
151 return int(value) if value is not None else None
153 for field in spec.fields:
154 if not isinstance(field, models.ForeignKey):
155 continue
156 related_id = row.get(field.attname)
157 related_spec = get_history_model_spec(field.remote_field.model)
158 if related_id is None or related_spec is None: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true
159 continue
160 related_row = (
161 _base_queryset(related_spec.model)
162 .filter(pk=related_id)
163 .values(*related_spec.field_names)
164 .first()
165 )
166 if related_row is not None: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true
167 flowsheet_id = resolve_row_flowsheet_id(related_spec, related_row)
168 if flowsheet_id is not None:
169 return flowsheet_id
170 return None
173def resolve_row_flowsheet_state_id(
174 spec: HistoryModelSpec,
175 row: dict[str, Any],
176) -> int | None:
177 """Resolve the owning state from a normalized row snapshot."""
178 if "flowsheet_state_id" in row:
179 value = row.get("flowsheet_state_id")
180 return int(value) if value is not None else None
182 for field in spec.fields:
183 if not isinstance(field, models.ForeignKey):
184 continue
185 related_id = row.get(field.attname)
186 related_spec = get_history_model_spec(field.remote_field.model)
187 if related_id is None or related_spec is None: 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true
188 continue
189 related_row = (
190 _base_queryset(related_spec.model)
191 .filter(pk=related_id)
192 .values(*related_spec.field_names)
193 .first()
194 )
195 if related_row is not None:
196 state_id = resolve_row_flowsheet_state_id(related_spec, related_row)
197 if state_id is not None: 197 ↛ 182line 197 didn't jump to line 182 because the condition on line 197 was always true
198 return state_id
199 return None
202def resolve_instance_flowsheet_id(instance: models.Model) -> int | None:
203 """Resolve and validate the direct or through-table flowsheet owner."""
204 spec = get_history_model_spec(type(instance))
205 if spec is None: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true
206 return None
207 row = {field.attname: getattr(instance, field.attname) for field in spec.fields}
208 return resolve_row_flowsheet_id(spec, row)
211def resolve_instance_flowsheet_state_id(instance: models.Model) -> int | None:
212 """Resolve the owning flowsheet state for a tracked model instance."""
213 spec = get_history_model_spec(type(instance))
214 if spec is None: 214 ↛ 215line 214 didn't jump to line 215 because the condition on line 214 was never true
215 return None
216 row = {field.attname: getattr(instance, field.attname) for field in spec.fields}
217 return resolve_row_flowsheet_state_id(spec, row)
220def _property_owner_id(property_info_id: int | None) -> int | None:
221 if property_info_id is None: 221 ↛ 222line 221 didn't jump to line 222 because the condition on line 221 was never true
222 return None
223 from core.auxiliary.models.PropertyInfo import PropertyInfo
225 return (
226 PropertyInfo._base_manager.filter(pk=property_info_id)
227 .values_list("set__simulationObject_id", flat=True)
228 .first()
229 )
232def _property_value_owner_id(property_value_id: int | None) -> int | None:
233 if property_value_id is None: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 return None
235 from core.auxiliary.models.PropertyValue import PropertyValue
237 return (
238 PropertyValue._base_manager.filter(pk=property_value_id)
239 .values_list("property__set__simulationObject_id", flat=True)
240 .first()
241 )
244def affected_ids_for_instance(instance: models.Model) -> tuple[set[int], set[int]]:
245 """Derive canvas object/group identities from one changed tracked row."""
246 label = instance._meta.label_lower.lower()
247 object_ids: set[int] = set()
248 group_ids: set[int] = set()
250 def add_object(value: int | None) -> None:
251 if value is not None:
252 object_ids.add(int(value))
254 def add_group(value: int | None) -> None:
255 if value is not None:
256 group_ids.add(int(value))
258 if label == "flowsheetinternals_unitops.simulationobject":
259 add_object(instance.pk)
260 elif label == "flowsheetinternals_unitops.port":
261 add_object(instance.unitOp_id)
262 add_object(instance.stream_id)
263 elif label == "flowsheetinternals_graphicdata.graphicobject":
264 add_object(instance.simulationObject_id)
265 add_group(instance.group_id)
266 elif label == "flowsheetinternals_graphicdata.grouping":
267 add_object(instance.simulationObject_id)
268 add_group(instance.pk)
269 elif label == "flowsheetinternals_graphicdata.portanchorplacement":
270 add_group(instance.grouping_id)
271 port = getattr(instance, "port", None)
272 if port is not None: 272 ↛ 322line 272 didn't jump to line 322 because the condition on line 272 was always true
273 add_object(port.unitOp_id)
274 add_object(port.stream_id)
275 elif label == "core_auxiliary.propertyset":
276 add_object(instance.simulationObject_id)
277 elif label == "core_auxiliary.propertyinfo":
278 if instance.set_id is not None: 278 ↛ 322line 278 didn't jump to line 322 because the condition on line 278 was always true
279 add_object(
280 instance.set.simulationObject_id
281 if getattr(instance, "set", None) is not None
282 else None
283 )
284 elif label == "core_auxiliary.propertyvalue":
285 add_object(_property_owner_id(instance.property_id))
286 elif label == "core_auxiliary.indexeditem":
287 add_object(instance.owner_id)
288 elif label == "core_auxiliary.controlvalue":
289 add_object(_property_value_owner_id(instance.manipulated_id))
290 add_object(_property_value_owner_id(instance.setPoint_id))
291 elif label == "core_auxiliary.recycledata":
292 add_object(instance.simulationObject_id)
293 add_object(instance.tearObject_id)
294 elif label == "core_auxiliary.recycleproperty":
295 add_object(_property_owner_id(instance.propertyInfo_id))
296 elif label == "pinchanalysis.streamdataentry":
297 add_object(instance.unitop_id)
298 add_group(instance.group_id)
299 elif label == "pinchanalysis.segment":
300 entry = getattr(instance, "stream_data_entry", None)
301 if entry is not None: 301 ↛ 322line 301 didn't jump to line 322 because the condition on line 301 was always true
302 add_object(entry.unitop_id)
303 add_group(entry.group_id)
304 elif spec := get_history_model_spec(type(instance)): 304 ↛ 322line 304 didn't jump to line 322 because the condition on line 304 was always true
305 if spec.is_automatic_through_model: 305 ↛ 322line 305 didn't jump to line 322 because the condition on line 305 was always true
306 for field in spec.fields:
307 if not isinstance(field, models.ForeignKey):
308 continue
309 related_id = getattr(instance, field.attname)
310 if related_id is None: 310 ↛ 311line 310 didn't jump to line 311 because the condition on line 310 was never true
311 continue
312 related = (
313 _base_queryset(field.remote_field.model)
314 .filter(pk=related_id)
315 .first()
316 )
317 if related is not None: 317 ↛ 306line 317 didn't jump to line 306 because the condition on line 317 was always true
318 related_objects, related_groups = affected_ids_for_instance(related)
319 object_ids.update(related_objects)
320 group_ids.update(related_groups)
322 return object_ids, group_ids