Coverage for backend/django/core/auxiliary/models/Flowsheet.py: 74%
103 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.db import connection, models, transaction
2from django.db.utils import NotSupportedError
3from django.db.models import Prefetch, QuerySet
4from django.utils import timezone
6from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType
7from authentication.user.models import User
8from authentication.user.AccessTable import AccessTable
9import core.auxiliary.enums.ViewType as ViewType
10from core.auxiliary.on_delete import (
11 protect_current_state_unless_flowsheet_deleting,
12)
15class Flowsheet(models.Model):
16 name = models.CharField(max_length=64, default="Flowsheet")
17 project = models.ForeignKey(
18 "Project", on_delete=models.CASCADE, related_name="flowsheets", null=True
19 )
20 savedDate = models.DateTimeField(null=True)
21 owner = models.ForeignKey(
22 User, on_delete=models.CASCADE, related_name="flowsheets", null=True
23 )
24 flowsheet_template_type = models.CharField(
25 max_length=32,
26 choices=FlowsheetTemplateType.choices,
27 default=FlowsheetTemplateType.NotTemplate,
28 )
29 current_state = models.OneToOneField(
30 "FlowsheetState",
31 on_delete=protect_current_state_unless_flowsheet_deleting,
32 related_name="current_for_flowsheet",
33 )
34 revision_sequence = models.PositiveBigIntegerField(default=0)
35 content_revision = models.PositiveBigIntegerField(default=0)
36 auto_snapshot_after_single_solve = models.BooleanField(default=False)
37 created_at = models.DateTimeField(auto_now_add=True)
38 binned_at = models.DateTimeField(null=True)
40 @classmethod
41 @transaction.atomic
42 def create(cls, **kwargs):
43 from flowsheetInternals.graphicData.models.groupingModel import Grouping
44 from PinchAnalysis.models.StreamDataProject import StreamDataProject
45 from core.auxiliary.models.FlowsheetState import FlowsheetState
47 saved_date = timezone.now()
48 explicit_name = kwargs.get("name")
49 project = kwargs.get("project")
51 field_values = {
52 "name": explicit_name or cls.next_default_name(project=project),
53 "project": project,
54 "savedDate": saved_date,
55 "owner": kwargs.get("owner"),
56 "flowsheet_template_type": kwargs.get(
57 "flowsheet_template_type",
58 FlowsheetTemplateType.NotTemplate,
59 ),
60 "auto_snapshot_after_single_solve": kwargs.get(
61 "auto_snapshot_after_single_solve",
62 False,
63 ),
64 }
66 # PostgreSQL foreign keys are deferred by Django. Reserve the state
67 # primary key so both sides of the required stable/state cycle can be
68 # inserted in one transaction without ever committing a null pointer.
69 if connection.vendor != "postgresql": 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true
70 raise NotSupportedError(
71 "Flowsheet construction requires PostgreSQL deferred foreign keys."
72 )
73 with connection.cursor() as cursor:
74 cursor.execute(
75 "SELECT nextval(pg_get_serial_sequence(%s, %s))",
76 [FlowsheetState._meta.db_table, FlowsheetState._meta.pk.column],
77 )
78 state_id = cursor.fetchone()[0]
80 instance = Flowsheet.objects.create(
81 **field_values,
82 current_state_id=state_id,
83 )
84 state = FlowsheetState(
85 pk=state_id,
86 flowsheet=instance,
87 created_by=instance.owner,
88 build_version=kwargs.get("buildVersion", "Not set"),
89 build_date=kwargs.get("buildDate", "No Build Date Set"),
90 source_saved_at=saved_date,
91 )
92 state.save(force_insert=True)
94 if not kwargs.get("initialize_state", True):
95 # Clone/restore services populate the aggregate before their outer
96 # transaction commits. Public creation retains the full factory.
97 return instance
99 # create the root group
100 rootGroup = Grouping.create(
101 state,
102 group=None,
103 componentName="Flowsheet",
104 visible=True,
105 isRoot=True,
106 )
108 state.root_grouping = rootGroup
109 state.save(update_fields=["root_grouping"])
111 StreamDataProject.create(rootGroup, flowsheet_state=state)
112 return instance
114 def clean(self):
115 """Require the selected working state to belong to this stable flowsheet."""
117 super().clean()
118 if self.current_state_id is None: 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true
119 return
120 if self.current_state.flowsheet_id != self.pk: 120 ↛ 126line 120 didn't jump to line 126 because the condition on line 120 was always true
121 from django.core.exceptions import ValidationError
123 raise ValidationError(
124 {"current_state": "The current state must belong to this flowsheet."}
125 )
126 if not self.current_state.is_working:
127 from django.core.exceptions import ValidationError
129 raise ValidationError(
130 {"current_state": "The current state must be working."}
131 )
133 @classmethod
134 def next_default_name(cls, *, project=None) -> str:
135 """
136 Return the next generated flowsheet name within a project.
138 Explicitly named flowsheets do not advance the generated-name sequence;
139 only names already matching ``Flowsheet-x`` are considered.
140 """
141 queryset = cls.objects.filter(project=project)
142 max_index = 0
143 for name in queryset.values_list("name", flat=True):
144 prefix = "Flowsheet-"
145 if not name or not name.startswith(prefix):
146 continue
148 suffix = name.removeprefix(prefix)
149 if suffix.isdecimal(): 149 ↛ 143line 149 didn't jump to line 143 because the condition on line 149 was always true
150 max_index = max(max_index, int(suffix))
151 return f"Flowsheet-{max_index + 1}"
153 def set_saved_date(self):
154 """Timestamp fresh database state without changing active selection."""
156 from core.auxiliary.services.flowsheet_template_transitions import (
157 touch_flowsheet_saved_date,
158 )
160 flowsheet = touch_flowsheet_saved_date(
161 candidate=self,
162 activate_if_regular=False,
163 )
164 self.savedDate = flowsheet.savedDate
166 def record_content_change(self):
167 """Record an unjournalled content change and invalidate undo/redo."""
169 from flowsheetInternals.unitops.services.edit_operations.context import (
170 history_transition_in_progress,
171 )
173 if history_transition_in_progress():
174 return
176 from core.auxiliary.services.flowsheet_template_transitions import (
177 touch_flowsheet_saved_date,
178 )
180 with transaction.atomic():
181 flowsheet = touch_flowsheet_saved_date(
182 candidate=self,
183 activate_if_regular=True,
184 )
185 flowsheet.invalidate_edit_history()
186 self.content_revision = flowsheet.content_revision
187 self.savedDate = flowsheet.savedDate
188 self.flowsheet_template_type = flowsheet.flowsheet_template_type
190 def invalidate_edit_history(self) -> None:
191 """Advance content identity and invalidate edits for locked content.
193 Callers must hold this flowsheet's row lock. This lower-level helper is
194 also used when revision restore has already taken the lock and replaced
195 the complete working state.
196 """
198 from flowsheetInternals.unitops.models.FlowsheetEditOperation import (
199 FlowsheetEditOperation,
200 )
202 self.content_revision += 1
203 self.save(update_fields=["content_revision"])
204 # Once the aggregate changes outside the recorder, none of these
205 # payloads can be replayed safely, so retaining them only wastes space.
206 FlowsheetEditOperation.objects.filter(flowsheet=self).delete()
208 @classmethod
209 def with_user_access(cls, queryset: QuerySet, user: User) -> QuerySet:
210 if user is None or not user.is_authenticated:
211 return queryset
213 return queryset.prefetch_related(
214 Prefetch(
215 "access_list",
216 queryset=AccessTable.objects.filter(user=user),
217 to_attr="current_user_access_entries",
218 )
219 )
221 @classmethod
222 def get_flowsheets_by_view_type(cls, user, view_type):
223 # filter to exclude templates
224 base_query = cls.objects.filter(
225 flowsheet_template_type=FlowsheetTemplateType.NotTemplate
226 )
228 if view_type == ViewType.ALL:
229 return base_query.filter(access_list__user=user)
230 elif view_type == ViewType.SHARED:
231 return base_query.filter(access_list__user=user).exclude(owner=user)
232 elif view_type == ViewType.OWNED:
233 return base_query.filter(owner=user)
234 else:
235 return base_query.filter(owner=user)