Coverage for backend/django/flowsheetInternals/unitops/services/edit_operations/mixins.py: 87%
69 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.core.exceptions import ImproperlyConfigured
2from rest_framework import status
3from rest_framework.response import Response
5from core.auxiliary.models.Flowsheet import Flowsheet
6from core.viewset import ModelViewSet
7from flowsheetInternals.unitops.models.FlowsheetEditOperation import (
8 FlowsheetEditOperation,
9)
11from .mutation import flowsheet_edit
12from .scope import get_history_model_spec
15class FlowsheetHistoryMixin:
16 """Shared configuration and ownership checks for history-aware CRUD."""
18 history_label: str
19 history_kind: str = FlowsheetEditOperation.Kind.FieldPatch
20 history_create_label: str | None = None
21 history_update_label: str | None = None
22 history_destroy_label: str | None = None
24 def get_history_label(self, action: str) -> str:
25 """Return an optional action-specific label or the shared label."""
26 label = getattr(self, f"history_{action}_label", None) or getattr(
27 self, "history_label", None
28 )
29 if not label: 29 ↛ 30line 29 didn't jump to line 30 because the condition on line 29 was never true
30 raise ImproperlyConfigured(
31 f"{type(self).__name__} must declare history_label."
32 )
33 return label
35 def validate_history_model(self, model) -> None:
36 """Fail loudly if a history viewset targets an excluded model."""
37 spec = get_history_model_spec(model)
38 if spec is None or spec.scope != "primary": 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true
39 raise ImproperlyConfigured(
40 f"{model._meta.label} is not a primary flowsheet history model."
41 )
43 def get_history_flowsheet(self, instance) -> Flowsheet:
44 """Return the stable owner of the row's flowsheet state."""
45 self.validate_history_model(type(instance))
46 return instance.flowsheet_state.flowsheet
49class FlowsheetHistoryCreateMixin(FlowsheetHistoryMixin):
50 """Give ordinary DRF model creation an automatic history boundary."""
52 def get_history_flowsheet_for_create(self, serializer) -> Flowsheet:
53 """Resolve ownership from validated data or the standard query context."""
54 model = serializer.Meta.model
55 self.validate_history_model(model)
56 flowsheet_state = serializer.validated_data.get("flowsheet_state")
57 if flowsheet_state is not None: 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true
58 return flowsheet_state.flowsheet
59 flowsheet_id = self.request.query_params.get("flowsheet")
60 if flowsheet_id is None: 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true
61 raise ImproperlyConfigured(
62 "A history-enabled create must supply its flowsheet through the "
63 "serializer or the standard flowsheet query parameter."
64 )
65 return Flowsheet.objects.get(pk=flowsheet_id)
67 def create(self, request, *args, **kwargs):
68 serializer = self.get_serializer(data=request.data)
69 serializer.is_valid(raise_exception=True)
70 flowsheet = self.get_history_flowsheet_for_create(serializer)
71 with flowsheet_edit(
72 flowsheet=flowsheet,
73 user=request.user,
74 kind=self.history_kind,
75 label_key=self.get_history_label("create"),
76 ) as mutation:
77 # Pin ownership to the locked working state so stale or historical
78 # state identifiers can never be written through ordinary CRUD.
79 serializer.save(flowsheet_state=mutation.flowsheet.current_state)
80 response = Response(
81 serializer.data,
82 status=status.HTTP_201_CREATED,
83 headers=self.get_success_headers(serializer.data),
84 )
85 return mutation.add_to_response(response)
88class FlowsheetHistoryUpdateMixin(FlowsheetHistoryMixin):
89 """Give ordinary DRF model updates an automatic history boundary.
91 The model marker and recorder determine changed rows and affected canvas
92 identities; feature code supplies only the product-facing operation label.
93 """
95 def update(self, request, *args, **kwargs):
96 partial = kwargs.pop("partial", False)
97 candidate = self.get_object()
98 flowsheet = self.get_history_flowsheet(candidate)
99 with flowsheet_edit(
100 flowsheet=flowsheet,
101 user=request.user,
102 kind=self.history_kind,
103 label_key=self.get_history_label("update"),
104 ) as mutation:
105 # Reload after taking the flowsheet lock. A concurrent edit may have
106 # committed while this request waited, and ModelSerializer.save()
107 # writes every concrete field from its instance.
108 instance = self.get_object()
109 serializer = self.get_serializer(
110 instance,
111 data=request.data,
112 partial=partial,
113 )
114 serializer.is_valid(raise_exception=True)
115 # Ownership comes from the locked row, never from a writable request
116 # field. Mutating validated_data preserves custom perform_update hooks.
117 serializer.validated_data["flowsheet_state"] = (
118 mutation.flowsheet.current_state
119 )
120 self.perform_update(serializer)
122 if getattr(instance, "_prefetched_objects_cache", None): 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 instance._prefetched_objects_cache = {}
124 response = Response(serializer.data)
125 return mutation.add_to_response(response)
128class FlowsheetHistoryDestroyMixin(FlowsheetHistoryMixin):
129 """Give ordinary DRF model deletion an automatic history boundary."""
131 def destroy(self, request, *args, **kwargs):
132 instance = self.get_object()
133 flowsheet = self.get_history_flowsheet(instance)
134 with flowsheet_edit(
135 flowsheet=flowsheet,
136 user=request.user,
137 kind=self.history_kind,
138 label_key=self.get_history_label("destroy"),
139 ) as mutation:
140 self.perform_destroy(instance)
141 return mutation.add_to_response(Response(status=status.HTTP_204_NO_CONTENT))
144class FlowsheetMutationViewSet(
145 FlowsheetHistoryCreateMixin,
146 FlowsheetHistoryUpdateMixin,
147 FlowsheetHistoryDestroyMixin,
148 ModelViewSet,
149):
150 """Standard CRUD base for a primary flowsheet history model.
152 Feature code declares its serializer, queryset, and product label; the
153 platform supplies atomic create/update/delete history automatically.
154 """
156 pass