Coverage for backend/django/Economics/studies/services/mutation_api.py: 86%
91 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""API workflow helpers for study mutation endpoints."""
3from __future__ import annotations
5from django.db import transaction
6from rest_framework import status
7from rest_framework.exceptions import NotFound, ValidationError
9from core.auxiliary.models.PropertyInfo import PropertyInfo
10from Economics.costing.capital.generated_lines import sync_generated_capital_lines
11from Economics.costing.costable_items.items import enable_costing_for_simulation_object
12from Economics.costing.costable_items.serializers import CostableItemSerializer
13from Economics.costing.operating.serializers import (
14 OperatingCostLineSerializer,
15 OperatingLinesFromPropertiesRequestSerializer,
16 OperatingLineFromPropertyRequestSerializer,
17 OperatingStreamPropertyOptionSerializer,
18)
19from Economics.costing.operating.stream_properties import (
20 create_operating_line_from_output_property,
21 output_stream_property_options,
22)
23from Economics.results.serializers import EconomicsResultRunSerializer, RecalculateRequestSerializer
24from Economics.results.services.lifecycle.lifecycle import recalculate_presentation_results
25from Economics.shared.access import require_write_access
26from Economics.studies.models import EconomicsStudy
27from Economics.studies.serializers import DuplicateStudyRequestSerializer, EnableCostingRequestSerializer
28from Economics.studies.services.baseline_access import require_readable_baseline_studies_for_user
29from Economics.studies.services.result_staleness import mark_study_stale
30from Economics.studies.services.study_copy import duplicate_study
31from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
34def operating_property_options_data(study: EconomicsStudy) -> list[dict]:
35 """Return operating stream property options for a study."""
37 return OperatingStreamPropertyOptionSerializer(output_stream_property_options(study), many=True).data
40def create_operating_line_from_property_response(
41 *,
42 study: EconomicsStudy,
43 user,
44 data,
45 serializer_context: dict,
46) -> tuple[dict, int]:
47 """Create one operating line from a selected output property."""
49 require_write_access(user)
50 request_serializer = OperatingLineFromPropertyRequestSerializer(data=data)
51 request_serializer.is_valid(raise_exception=True)
52 options_by_property = {
53 option.property_info: option for option in output_stream_property_options(study)
54 }
55 property_id = request_serializer.validated_data["property_info"]
56 try:
57 property_info = (
58 PropertyInfo.objects.select_related("set", "set__simulationObject")
59 .prefetch_related("values")
60 .get(pk=property_id)
61 )
62 except PropertyInfo.DoesNotExist as error:
63 raise NotFound("Selected property was not found.") from error
64 try:
65 source_option = options_by_property.get(property_id)
66 if source_option is None: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true
67 raise ValueError("Operating-line source must be a suggested current scalar operating property.")
68 line = create_operating_line_from_output_property(
69 study=study,
70 property_info=property_info,
71 category=request_serializer.validated_data["category"],
72 economic_effect=request_serializer.validated_data["economic_effect"],
73 outlet_stream_disposition=request_serializer.validated_data.get("outlet_stream_disposition", ""),
74 rate_type=request_serializer.validated_data.get("rate_type", None),
75 source_option=source_option,
76 )
77 except ValueError as error:
78 raise _bad_request_from_error(error) from error
79 mark_study_stale(study, reason="economics_operating_line_property_selected")
80 return OperatingCostLineSerializer(line, context=serializer_context).data, status.HTTP_201_CREATED
83def create_operating_lines_from_properties_response(
84 *,
85 study: EconomicsStudy,
86 user,
87 data,
88 serializer_context: dict,
89) -> tuple[list[dict], int]:
90 """Create operating lines from selected output properties."""
92 require_write_access(user)
93 request_serializer = OperatingLinesFromPropertiesRequestSerializer(data=data)
94 request_serializer.is_valid(raise_exception=True)
95 line_requests = request_serializer.validated_data["lines"]
96 property_infos = (
97 PropertyInfo.objects.select_related("set", "set__simulationObject")
98 .prefetch_related("values")
99 .in_bulk([line["property_info"] for line in line_requests])
100 )
101 missing_property_ids = [
102 line["property_info"]
103 for line in line_requests
104 if line["property_info"] not in property_infos
105 ]
106 if missing_property_ids: 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 raise NotFound("Selected property was not found.")
108 options_by_property = {
109 option.property_info: option for option in output_stream_property_options(study)
110 }
111 unavailable_property_ids = [
112 line["property_info"]
113 for line in line_requests
114 if line["property_info"] not in options_by_property
115 ]
116 if unavailable_property_ids:
117 raise _bad_request_from_error(
118 ValueError("Operating-line source must be a suggested current scalar operating property.")
119 )
120 try:
121 with transaction.atomic():
122 lines = [
123 create_operating_line_from_output_property(
124 study=study,
125 property_info=property_infos[line_request["property_info"]],
126 category=line_request["category"],
127 economic_effect=line_request["economic_effect"],
128 outlet_stream_disposition=line_request.get("outlet_stream_disposition", ""),
129 rate_type=line_request.get("rate_type", None),
130 source_option=options_by_property[line_request["property_info"]],
131 )
132 for line_request in line_requests
133 ]
134 except ValueError as error:
135 raise _bad_request_from_error(error) from error
136 mark_study_stale(study, reason="economics_operating_line_property_selected")
137 return OperatingCostLineSerializer(lines, many=True, context=serializer_context).data, status.HTTP_201_CREATED
140def enable_costing_response(
141 *,
142 study: EconomicsStudy,
143 user,
144 data,
145 serializer_context: dict,
146) -> tuple[dict, int]:
147 """Enable costing for one simulation object and return the costable item payload."""
149 require_write_access(user)
150 serializer = EnableCostingRequestSerializer(data=data)
151 serializer.is_valid(raise_exception=True)
152 try:
153 simulation_object = SimulationObject.objects.get(pk=serializer.validated_data["simulation_object"])
154 result = enable_costing_for_simulation_object(study=study, simulation_object=simulation_object)
155 except (SimulationObject.DoesNotExist, ValueError) as exc:
156 raise _bad_request_from_error(exc) from exc
158 sync_generated_capital_lines(study)
159 mark_study_stale(study, reason="economics_enable_costing_api", requires_solve=True)
160 response = CostableItemSerializer(result.costable_item, context=serializer_context).data
161 response["supported"] = result.supported
162 return response, status.HTTP_200_OK
165def duplicate_study_instance(*, source: EconomicsStudy, user, data) -> EconomicsStudy:
166 """Duplicate a study after validating the request payload."""
168 require_write_access(user)
169 serializer = DuplicateStudyRequestSerializer(data=data)
170 serializer.is_valid(raise_exception=True)
171 return duplicate_study(source, requested_name=serializer.validated_data.get("name", ""))
174def recalculate_data(
175 *,
176 study: EconomicsStudy,
177 user,
178 data,
179 serializer_context: dict,
180) -> dict:
181 """Validate and run a synchronous presentation-result recalculation."""
183 require_write_access(user)
184 require_readable_baseline_studies_for_user(study, user=user)
185 serializer = RecalculateRequestSerializer(data=data)
186 serializer.is_valid(raise_exception=True)
187 try:
188 result_run = recalculate_presentation_results(study, reason=serializer.validated_data["reason"])
189 except ValueError as exc:
190 raise _bad_request_from_error(exc) from exc
191 return EconomicsResultRunSerializer(result_run, context=serializer_context).data
194def _bad_request_from_error(error: Exception) -> ValidationError:
195 """Convert a domain validation exception into a DRF bad-request payload."""
197 return ValidationError({"detail": str(error)})