Coverage for backend/django/Economics/studies/services/project_scope.py: 84%

1"""Project-scoped Economics lookup helpers. 

2 

3Economics studies stay owned by a single flowsheet, but project-level workflows 

4need to discover sibling flowsheets without letting unrelated readable projects 

5leak into study creation, baseline, or comparison paths. 

6""" 

7 

8from django.db.models import Case, IntegerField, Prefetch, QuerySet, Value, When 

9from rest_framework.exceptions import PermissionDenied, ValidationError 

10 

11from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

12from core.auxiliary.models.Flowsheet import Flowsheet 

13from core.auxiliary.models.Project import Project 

14from core.managers import has_flowsheet_read_access, has_flowsheet_write_access, project_access_filter 

15from core.validation import get_current_flowsheet 

16from Economics.studies.models import EconomicsStudy 

17from Economics.results.models import EconomicsResultDependency, EconomicsResultRun 

18 

19 

20def current_flowsheet_from_context(user) -> Flowsheet: 

21 """Resolve and authorize the request-context flowsheet.""" 

22 

23 context = get_current_flowsheet() or {} 

24 flowsheet_id = context.get("flowsheet") 

25 if flowsheet_id is None: 

26 raise ValidationError({"flowsheet": "A flowsheet context is required."}) 

27 

28 flowsheet = ( 

29 Flowsheet.objects 

30 .select_related("project") 

31 .filter(pk=flowsheet_id) 

32 .first() 

33 ) 

34 if flowsheet is None: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 raise ValidationError({"flowsheet": "The selected flowsheet was not found."}) 

36 if not has_flowsheet_read_access(user, flowsheet.pk): 

37 raise PermissionDenied("User does not have access to this flowsheet.") 

38 if flowsheet.project_id is None: 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true

39 raise ValidationError({"flowsheet": "The selected flowsheet is not part of a project."}) 

40 return flowsheet 

41 

42 

43def current_project_from_flowsheet_context(user) -> Project: 

44 """Return the project owned by the authorized request-context flowsheet.""" 

45 

46 return current_flowsheet_from_context(user).project 

47 

48 

49def project_non_template_flowsheets( 

50 project: Project, 

51 user, 

52 *, 

53 write: bool = False, 

54) -> QuerySet[Flowsheet]: 

55 """Return current-project flowsheets that are valid Economics candidates.""" 

56 

57 queryset = ( 

58 project.flowsheets 

59 .filter(flowsheet_template_type=FlowsheetTemplateType.NotTemplate) 

60 ) 

61 active_id = project.active_flowsheet_id 

62 return ( 

63 queryset 

64 .filter(project_access_filter(user, write=write)) 

65 .annotate( 

66 active_sort=Case( 

67 When(pk=active_id, then=Value(0)), 

68 default=Value(1), 

69 output_field=IntegerField(), 

70 ) 

71 ) 

72 .distinct() 

73 .order_by("active_sort", "created_at", "id") 

74 ) 

75 

76 

77def resolve_target_study_flowsheet( 

78 context_flowsheet_id: int, 

79 target_flowsheet_id: int | None, 

80 user, 

81) -> Flowsheet: 

82 """Resolve a create-time target flowsheet inside the current project.""" 

83 

84 target_id = context_flowsheet_id if target_flowsheet_id is None else target_flowsheet_id 

85 context_flowsheet = ( 

86 Flowsheet.objects 

87 .select_related("project") 

88 .filter(pk=context_flowsheet_id) 

89 .first() 

90 ) 

91 if context_flowsheet is None or context_flowsheet.project_id is None: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true

92 raise ValidationError({"flowsheet": "The selected flowsheet was not found."}) 

93 if not has_flowsheet_read_access(user, context_flowsheet.pk): 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true

94 raise PermissionDenied("User does not have access to this flowsheet.") 

95 

96 readable_target = project_non_template_flowsheets( 

97 context_flowsheet.project, 

98 user, 

99 ).filter(pk=target_id).first() 

100 if readable_target is None: 

101 raise ValidationError( 

102 { 

103 "target_flowsheet": ( 

104 "Select a non-template flowsheet in the current project." 

105 ) 

106 } 

107 ) 

108 

109 target_flowsheet = project_non_template_flowsheets( 

110 context_flowsheet.project, 

111 user, 

112 write=True, 

113 ).filter(pk=target_id).first() 

114 if target_flowsheet is None: 

115 if not has_flowsheet_write_access(user, readable_target.pk): 115 ↛ 117line 115 didn't jump to line 117 because the condition on line 115 was always true

116 raise PermissionDenied("User does not have write access to this flowsheet.") 

117 raise ValidationError({"target_flowsheet": "Select an editable flowsheet."}) 

118 return target_flowsheet 

119 

120 

121def study_queryset_for_project(project: Project, user) -> QuerySet[EconomicsStudy]: 

122 """Load studies for every non-template flowsheet in a readable project.""" 

123 

124 current_state_ids = project_non_template_flowsheets(project, user).values( 

125 "current_state_id" 

126 ) 

127 active_id = project.active_flowsheet_id 

128 return ( 

129 EconomicsStudy._base_manager 

130 .select_related( 

131 "flowsheet_state__flowsheet", 

132 "settings_profile", 

133 "schedule_scenario", 

134 ) 

135 .filter(flowsheet_state_id__in=current_state_ids) 

136 .prefetch_related( 

137 Prefetch( 

138 "result_runs", 

139 queryset=EconomicsResultRun._base_manager.filter( 

140 flowsheet_state_id__in=current_state_ids, 

141 ).order_by("-created_at", "-pk") 

142 .prefetch_related( 

143 Prefetch( 

144 "dependencies", 

145 queryset=EconomicsResultDependency._base_manager.filter( 

146 flowsheet_state_id__in=current_state_ids, 

147 ), 

148 ) 

149 )[:1], 

150 to_attr="_economics_latest_result_runs", 

151 ) 

152 ) 

153 .annotate( 

154 active_flowsheet_sort=Case( 

155 When(flowsheet_state__flowsheet_id=active_id, then=Value(0)), 

156 default=Value(1), 

157 output_field=IntegerField(), 

158 ) 

159 ) 

160 .order_by( 

161 "active_flowsheet_sort", 

162 "flowsheet_state__flowsheet__name", 

163 "created_at", 

164 "id", 

165 ) 

166 ) 

167 

168 

169def study_is_in_current_project(study: EconomicsStudy, context_flowsheet_id: int) -> bool: 

170 """Return whether study belongs to a non-template sibling of context flowsheet.""" 

171 

172 context_flowsheet = Flowsheet.objects.filter(pk=context_flowsheet_id).only("project_id").first() 

173 if context_flowsheet is None or context_flowsheet.project_id is None: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true

174 return False 

175 flowsheet = study.flowsheet_state.flowsheet 

176 return ( 

177 flowsheet.project_id == context_flowsheet.project_id 

178 and flowsheet.flowsheet_template_type == FlowsheetTemplateType.NotTemplate 

179 )