Coverage for backend/django/core/auxiliary/services/flowsheet_states/revisions.py: 95%

1"""Transactional revision creation, retention, listing, and restore services.""" 

2 

3from dataclasses import dataclass, field 

4from enum import StrEnum 

5 

6from django.db import transaction 

7from django.utils import timezone 

8 

9from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

10from core.auxiliary.enums.generalEnums import TaskStatus 

11from core.auxiliary.models.Flowsheet import Flowsheet 

12from core.auxiliary.models.FlowsheetState import ( 

13 FlowsheetRevisionKind, 

14 FlowsheetState, 

15 FlowsheetStateRole, 

16) 

17from core.auxiliary.models.Task import Task 

18 

19from .cloning import CloneResult, clone_state 

20from .registry import CloneProfile 

21from .validation import repair_legacy_formula_unit_references 

22 

23 

24MAX_RETAINED_REVISIONS = 100 

25ACTIVE_TASK_STATUSES = ( 

26 TaskStatus.Pending, 

27 TaskStatus.Running, 

28 TaskStatus.Cancelling, 

29) 

30ACTIVE_TASKS_ERROR_CODE = "active_tasks" 

31ACTIVE_TASKS_ERROR_MESSAGE = ( 

32 "Let active tasks finish or cancel them before saving or restoring a revision." 

33) 

34CLEARED_CALCULATED_DATA_CATEGORIES = ( 

35 "scenario_solutions", 

36 "economics_calculations", 

37 "generated_pinch_outputs", 

38 "diagnostics", 

39 "pgraph_paths", 

40 "result_summary_caches", 

41) 

42 

43 

44class FlowsheetRevisionCategory(StrEnum): 

45 """Product categories available when browsing retained revisions.""" 

46 

47 ALL = "all" 

48 MANUAL = "manual" 

49 SOLVE = "solve" 

50 

51 

52class RevisionOperationError(ValueError): 

53 """Raised when revision lifecycle preconditions are not satisfied.""" 

54 

55 

56class ActiveTasksError(RevisionOperationError): 

57 """Raised when a revision operation conflicts with active flowsheet work.""" 

58 

59 

60@dataclass 

61class RestoreResult: 

62 """Deterministic data returned after replacing a working state.""" 

63 

64 flowsheet: Flowsheet 

65 current_state: FlowsheetState 

66 warnings: list[dict] = field(default_factory=list) 

67 cleared_calculated_data_categories: tuple[str, ...] = ( 

68 CLEARED_CALCULATED_DATA_CATEGORIES 

69 ) 

70 

71 

72def _require_revision_enabled(flowsheet: Flowsheet) -> None: 

73 """Reject revision operations for template flowsheets in v1.""" 

74 

75 if flowsheet.flowsheet_template_type != FlowsheetTemplateType.NotTemplate: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 raise RevisionOperationError("Revision history is not available for templates.") 

77 

78 

79def has_active_tasks(*, flowsheet_id: int) -> bool: 

80 """Return whether any non-terminal task belongs to the stable flowsheet.""" 

81 

82 return Task.objects.filter( 

83 flowsheet_id=flowsheet_id, 

84 status__in=ACTIVE_TASK_STATUSES, 

85 ).exists() 

86 

87 

88def _require_no_active_tasks(flowsheet: Flowsheet) -> None: 

89 """Reject working-state copies and replacements while tasks are active.""" 

90 

91 if has_active_tasks(flowsheet_id=flowsheet.pk): 

92 raise ActiveTasksError(ACTIVE_TASKS_ERROR_MESSAGE) 

93 

94 

95def _prune_revisions_locked(flowsheet: Flowsheet) -> list[int]: 

96 """Synchronously delete immutable states beyond the newest retained 100.""" 

97 

98 stale_ids = list( 

99 FlowsheetState.objects.filter( 

100 flowsheet=flowsheet, 

101 role=FlowsheetStateRole.REVISION, 

102 ) 

103 .order_by("-revision_number") 

104 .values_list("pk", flat=True)[MAX_RETAINED_REVISIONS:] 

105 ) 

106 if stale_ids: 

107 FlowsheetState.objects.filter(pk__in=stale_ids).delete() 

108 return stale_ids 

109 

110 

111def _create_revision_locked( 

112 *, 

113 flowsheet: Flowsheet, 

114 kind: FlowsheetRevisionKind, 

115 created_by, 

116 label: str | None = None, 

117 prune: bool = True, 

118) -> tuple[FlowsheetState, CloneResult]: 

119 """Clone the locked current state into one numbered immutable revision.""" 

120 

121 next_number = flowsheet.revision_sequence + 1 

122 source_state = FlowsheetState.objects.select_for_update().get( 

123 pk=flowsheet.current_state_id, 

124 flowsheet=flowsheet, 

125 role=FlowsheetStateRole.WORKING, 

126 ) 

127 if kind == FlowsheetRevisionKind.MANUAL and label is None: 

128 label = f"Revision {next_number}" 

129 

130 revision = FlowsheetState.objects.create( 

131 flowsheet=flowsheet, 

132 role=FlowsheetStateRole.REVISION, 

133 revision_number=next_number, 

134 revision_kind=kind, 

135 label=label, 

136 created_by=created_by, 

137 restored_from_revision_number=( 

138 source_state.restored_from_revision_number 

139 ), 

140 source_saved_at=flowsheet.savedDate, 

141 build_version=source_state.build_version, 

142 build_date=source_state.build_date, 

143 ) 

144 clone_result = clone_state( 

145 source_state=source_state, 

146 target_state=revision, 

147 profile=CloneProfile.REVISION, 

148 ) 

149 flowsheet.revision_sequence = next_number 

150 flowsheet.save(update_fields=["revision_sequence"]) 

151 if prune: 151 ↛ 153line 151 didn't jump to line 153 because the condition on line 151 was always true

152 _prune_revisions_locked(flowsheet) 

153 return revision, clone_result 

154 

155 

156@transaction.atomic 

157def create_revision( 

158 *, 

159 flowsheet_id: int, 

160 kind: FlowsheetRevisionKind, 

161 created_by, 

162 label: str | None = None, 

163 repair_legacy_formula_units: bool = False, 

164) -> FlowsheetState: 

165 """Create one immutable revision while serializing revision operations.""" 

166 

167 flowsheet = Flowsheet.objects.select_for_update().select_related( 

168 "current_state" 

169 ).get(pk=flowsheet_id) 

170 _require_revision_enabled(flowsheet) 

171 _require_no_active_tasks(flowsheet) 

172 if repair_legacy_formula_units: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true

173 repair_legacy_formula_unit_references( 

174 source_state=flowsheet.current_state, 

175 profile=CloneProfile.REVISION, 

176 ) 

177 revision, _ = _create_revision_locked( 

178 flowsheet=flowsheet, 

179 kind=FlowsheetRevisionKind(kind), 

180 created_by=created_by, 

181 label=label, 

182 ) 

183 return revision 

184 

185 

186def list_revisions( 

187 *, 

188 flowsheet: Flowsheet, 

189 category: FlowsheetRevisionCategory = FlowsheetRevisionCategory.ALL, 

190): 

191 """Return immutable revisions newest first without exposing internal roles.""" 

192 

193 _require_revision_enabled(flowsheet) 

194 category = FlowsheetRevisionCategory(category) 

195 revisions = ( 

196 FlowsheetState.objects.filter( 

197 flowsheet=flowsheet, 

198 role=FlowsheetStateRole.REVISION, 

199 ) 

200 .select_related("created_by") 

201 .order_by("-revision_number") 

202 ) 

203 if category == FlowsheetRevisionCategory.MANUAL: 

204 return revisions.filter(revision_kind=FlowsheetRevisionKind.MANUAL) 

205 if category == FlowsheetRevisionCategory.SOLVE: 

206 return revisions.filter(revision_kind=FlowsheetRevisionKind.AUTO_SOLVE) 

207 return revisions 

208 

209 

210@transaction.atomic 

211def rename_revision( 

212 *, 

213 flowsheet_id: int, 

214 revision_state_id: int, 

215 label: str, 

216) -> FlowsheetState: 

217 """Rename revision metadata without changing its immutable state content.""" 

218 

219 flowsheet = Flowsheet.objects.select_for_update().get(pk=flowsheet_id) 

220 _require_revision_enabled(flowsheet) 

221 try: 

222 revision = FlowsheetState.objects.select_for_update().get( 

223 pk=revision_state_id, 

224 flowsheet=flowsheet, 

225 role=FlowsheetStateRole.REVISION, 

226 ) 

227 except FlowsheetState.DoesNotExist as exc: 

228 raise RevisionOperationError( 

229 "The selected revision does not belong to this flowsheet." 

230 ) from exc 

231 

232 revision.label = label 

233 revision.save(update_fields=["label"]) 

234 return revision 

235 

236 

237@transaction.atomic 

238def delete_revision(*, flowsheet_id: int, revision_state_id: int) -> None: 

239 """Permanently delete one retained revision without reusing its number.""" 

240 

241 flowsheet = Flowsheet.objects.select_for_update().get(pk=flowsheet_id) 

242 _require_revision_enabled(flowsheet) 

243 try: 

244 revision = FlowsheetState.objects.select_for_update().get( 

245 pk=revision_state_id, 

246 flowsheet=flowsheet, 

247 role=FlowsheetStateRole.REVISION, 

248 ) 

249 except FlowsheetState.DoesNotExist as exc: 

250 raise RevisionOperationError( 

251 "The selected revision does not belong to this flowsheet." 

252 ) from exc 

253 

254 revision.delete() 

255 

256 

257@transaction.atomic 

258def restore_revision( 

259 *, 

260 flowsheet_id: int, 

261 revision_state_id: int, 

262 restored_by, 

263) -> RestoreResult: 

264 """Replace current content from an immutable revision under one row lock.""" 

265 

266 flowsheet = Flowsheet.objects.select_for_update().select_related( 

267 "current_state" 

268 ).get(pk=flowsheet_id) 

269 _require_revision_enabled(flowsheet) 

270 _require_no_active_tasks(flowsheet) 

271 try: 

272 selected_revision = FlowsheetState.objects.select_for_update().get( 

273 pk=revision_state_id, 

274 flowsheet=flowsheet, 

275 role=FlowsheetStateRole.REVISION, 

276 ) 

277 except FlowsheetState.DoesNotExist as exc: 

278 raise RevisionOperationError( 

279 "The selected revision does not belong to this flowsheet." 

280 ) from exc 

281 

282 replaced_state = FlowsheetState.objects.select_for_update().get( 

283 pk=flowsheet.current_state_id, 

284 flowsheet=flowsheet, 

285 role=FlowsheetStateRole.WORKING, 

286 ) 

287 staging = FlowsheetState.objects.create( 

288 flowsheet=flowsheet, 

289 role=FlowsheetStateRole.STAGING, 

290 created_by=restored_by, 

291 restored_from_revision_number=selected_revision.revision_number, 

292 source_saved_at=selected_revision.source_saved_at, 

293 build_version=selected_revision.build_version, 

294 build_date=selected_revision.build_date, 

295 ) 

296 clone_result = clone_state( 

297 source_state=selected_revision, 

298 target_state=staging, 

299 profile=CloneProfile.RESTORE, 

300 ) 

301 replaced_state.role = FlowsheetStateRole.RETIRING 

302 replaced_state.save(update_fields=["role"]) 

303 staging.role = FlowsheetStateRole.WORKING 

304 staging.save(update_fields=["role"]) 

305 flowsheet.current_state = staging 

306 flowsheet.savedDate = timezone.now() 

307 flowsheet.save(update_fields=["current_state", "savedDate"]) 

308 flowsheet.invalidate_edit_history() 

309 replaced_state.delete() 

310 

311 result = RestoreResult( 

312 flowsheet=flowsheet, 

313 current_state=staging, 

314 warnings=clone_result.warnings, 

315 ) 

316 return result