Coverage for backend/django/flowsheetInternals/unitops/services/edit_operations/service.py: 88%
59 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from collections.abc import Iterable
3from django.db import transaction
4from django.db.models import Q
5from django.utils import timezone
6from rest_framework.exceptions import ValidationError
8from core.auxiliary.models.Flowsheet import Flowsheet
9from flowsheetInternals.unitops.models.FlowsheetEditOperation import (
10 FlowsheetEditOperation,
11)
13from .replay import apply_operation_payload
14from .types import OperationTransition
17MAX_HISTORY_DEPTH = 100
18CURRENT_SCHEMA_VERSION = 2
21def _prune_locked_flowsheet_history(*, flowsheet_id: int) -> int:
22 """Delete unusable rows while the caller holds the flowsheet row lock."""
23 operations = FlowsheetEditOperation.objects.filter(flowsheet_id=flowsheet_id)
24 deleted_count, _ = operations.filter(
25 Q(state=FlowsheetEditOperation.State.Invalidated)
26 | ~Q(schema_version=CURRENT_SCHEMA_VERSION)
27 ).delete()
29 retained_ids = list(
30 operations.order_by("-revision").values_list("pk", flat=True)[
31 :MAX_HISTORY_DEPTH
32 ]
33 )
34 pruned_count, _ = operations.exclude(pk__in=retained_ids).delete()
35 return deleted_count + pruned_count
38@transaction.atomic
39def record_edit_operation(
40 *,
41 flowsheet: Flowsheet,
42 user,
43 kind: str,
44 label_key: str,
45 forward_payload: dict,
46 inverse_payload: dict,
47 affected_object_ids: Iterable[int] = (),
48 affected_group_ids: Iterable[int] = (),
49) -> FlowsheetEditOperation:
50 """Persist one committed user action and allocate its flowsheet revision."""
51 locked_flowsheet = Flowsheet.objects.select_for_update().get(pk=flowsheet.pk)
53 # A new edit after undo starts a new branch, so its abandoned redo payloads
54 # can never be used again and should not consume database storage.
55 FlowsheetEditOperation.objects.filter(
56 flowsheet=locked_flowsheet,
57 state=FlowsheetEditOperation.State.Undone,
58 ).delete()
60 locked_flowsheet.content_revision += 1
61 locked_flowsheet.savedDate = timezone.now()
62 locked_flowsheet.save(update_fields=["content_revision", "savedDate"])
63 operation = FlowsheetEditOperation.objects.create(
64 flowsheet=locked_flowsheet,
65 user=user if getattr(user, "is_authenticated", False) else None,
66 kind=kind,
67 label_key=label_key,
68 revision=locked_flowsheet.content_revision,
69 last_revision=locked_flowsheet.content_revision,
70 schema_version=CURRENT_SCHEMA_VERSION,
71 forward_payload=forward_payload,
72 inverse_payload=inverse_payload,
73 affected_object_ids=sorted(set(affected_object_ids)),
74 affected_group_ids=sorted(set(affected_group_ids)),
75 )
76 _prune_locked_flowsheet_history(flowsheet_id=locked_flowsheet.pk)
77 flowsheet.content_revision = locked_flowsheet.content_revision
78 flowsheet.savedDate = locked_flowsheet.savedDate
79 return operation
82def record_transition(
83 *,
84 flowsheet: Flowsheet,
85 user,
86 kind: str,
87 label_key: str,
88 transition: OperationTransition,
89) -> FlowsheetEditOperation:
90 """Persist the normalized output produced by the automatic recorder."""
91 return record_edit_operation(
92 flowsheet=flowsheet,
93 user=user,
94 kind=kind,
95 label_key=label_key,
96 forward_payload=transition.forward_payload,
97 inverse_payload=transition.inverse_payload,
98 affected_object_ids=transition.affected_object_ids,
99 affected_group_ids=transition.affected_group_ids,
100 )
103def operation_response(operation: FlowsheetEditOperation) -> dict:
104 """Serialize operation metadata shared by mutation and history responses."""
105 return {
106 "operation_id": operation.pk,
107 "label": operation.label_key,
108 "revision": operation.last_revision,
109 "state": operation.state,
110 "affected_object_ids": operation.affected_object_ids,
111 "affected_group_ids": operation.affected_group_ids,
112 }
115@transaction.atomic
116def transition_operation(
117 operation: FlowsheetEditOperation,
118 *,
119 undo: bool,
120 expected_revision: int,
121) -> FlowsheetEditOperation:
122 """Undo or redo one operation after validating stack order and state."""
123 locked_flowsheet = Flowsheet.objects.select_for_update().get(
124 pk=operation.flowsheet_id
125 )
126 try:
127 locked_operation = FlowsheetEditOperation.objects.select_for_update().get(
128 pk=operation.pk,
129 flowsheet=locked_flowsheet,
130 )
131 except FlowsheetEditOperation.DoesNotExist as error:
132 raise ValidationError({"detail": "This edit is no longer available."}) from error
133 if locked_flowsheet.content_revision != expected_revision:
134 raise ValidationError(
135 {"detail": "The flowsheet changed. Refresh its edit history and try again."}
136 )
137 if locked_operation.schema_version != CURRENT_SCHEMA_VERSION: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true
138 raise ValidationError({"detail": "This edit uses an obsolete history format."})
140 expected_state = (
141 FlowsheetEditOperation.State.Applied
142 if undo
143 else FlowsheetEditOperation.State.Undone
144 )
145 if locked_operation.state != expected_state: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 raise ValidationError({"detail": "This edit is no longer available."})
148 if (
149 undo
150 and FlowsheetEditOperation.objects.filter(
151 flowsheet=locked_flowsheet,
152 revision__gt=locked_operation.revision,
153 state=FlowsheetEditOperation.State.Applied,
154 ).exists()
155 ):
156 raise ValidationError({"detail": "Undo the latest edit first."})
157 if ( 157 ↛ 165line 157 didn't jump to line 165 because the condition on line 157 was never true
158 not undo
159 and FlowsheetEditOperation.objects.filter(
160 flowsheet=locked_flowsheet,
161 revision__lt=locked_operation.revision,
162 state=FlowsheetEditOperation.State.Undone,
163 ).exists()
164 ):
165 raise ValidationError({"detail": "Redo edits in their original order."})
167 apply_operation_payload(
168 locked_operation.inverse_payload if undo else locked_operation.forward_payload,
169 flowsheet=locked_flowsheet,
170 )
171 locked_flowsheet.content_revision += 1
172 locked_flowsheet.savedDate = timezone.now()
173 locked_flowsheet.save(update_fields=["content_revision", "savedDate"])
174 locked_operation.state = (
175 FlowsheetEditOperation.State.Undone
176 if undo
177 else FlowsheetEditOperation.State.Applied
178 )
179 locked_operation.last_revision = locked_flowsheet.content_revision
180 locked_operation.save(update_fields=["state", "last_revision", "updated_at"])
181 return locked_operation