Coverage for backend/django/flowsheetInternals/unitops/services/edit_operations/mutation.py: 88%

1from contextlib import contextmanager 

2from dataclasses import dataclass 

3from typing import Iterator 

4 

5from django.db import transaction 

6 

7from core.auxiliary.models.Flowsheet import Flowsheet 

8from core.auxiliary.services.flowsheet_template_transitions import ( 

9 lock_flowsheet_content_mutation, 

10) 

11from flowsheetInternals.unitops.models.FlowsheetEditOperation import ( 

12 FlowsheetEditOperation, 

13) 

14 

15from .context import active_recorder, get_active_recorder 

16from .recorder import FlowsheetChangeRecorder 

17from .service import operation_response, record_transition 

18 

19 

20@dataclass 

21class FlowsheetMutation: 

22 """Result and locked context for one history-aware domain mutation.""" 

23 

24 flowsheet: Flowsheet 

25 operation: FlowsheetEditOperation | None = None 

26 

27 def add_to_response(self, response, *, header_only: bool = False): 

28 """Attach operation metadata without changing an endpoint's data contract.""" 

29 if self.operation is None: 

30 return response 

31 if ( 

32 header_only 

33 or response.status_code == 204 

34 or not isinstance(response.data, dict) 

35 ): 

36 response["X-Flowsheet-Edit-Operation"] = self.operation.pk 

37 else: 

38 response.data.update(operation_response(self.operation)) 

39 return response 

40 

41 def add_to_data(self, data: dict) -> dict: 

42 """Return response data enriched with operation metadata when present.""" 

43 if self.operation is None: 

44 return data 

45 return {**data, **operation_response(self.operation)} 

46 

47 

48@contextmanager 

49def flowsheet_edit( 

50 *, 

51 flowsheet: Flowsheet, 

52 user, 

53 kind: str, 

54 label_key: str, 

55) -> Iterator[FlowsheetMutation]: 

56 """Record all scoped database writes made by one flowsheet user action. 

57 

58 Nested domain services join the active recorder. Only the outer boundary 

59 allocates a revision and persists an operation. 

60 """ 

61 existing_recorder = get_active_recorder() 

62 if existing_recorder is not None: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 if ( 

64 existing_recorder.flowsheet_id != flowsheet.pk 

65 or existing_recorder.flowsheet_state_id != flowsheet.current_state_id 

66 ): 

67 raise ValueError("A flowsheet edit cannot span multiple flowsheets.") 

68 yield FlowsheetMutation(flowsheet=flowsheet) 

69 return 

70 

71 with transaction.atomic(): 

72 locked_flowsheet = lock_flowsheet_content_mutation(candidate=flowsheet) 

73 mutation = FlowsheetMutation(flowsheet=locked_flowsheet) 

74 recorder = FlowsheetChangeRecorder( 

75 locked_flowsheet.pk, 

76 locked_flowsheet.current_state_id, 

77 kind=kind, 

78 ) 

79 token = active_recorder.set(recorder) 

80 try: 

81 yield mutation 

82 transition = recorder.build_transition() 

83 if transition is not None: 

84 mutation.operation = record_transition( 

85 flowsheet=locked_flowsheet, 

86 user=user, 

87 kind=kind, 

88 label_key=label_key, 

89 transition=transition, 

90 ) 

91 finally: 

92 active_recorder.reset(token)