Coverage for backend/django/core/auxiliary/views/DownloadMSSData.py: 73%
130 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-05-13 02:47 +0000
1from core.auxiliary.models.DataColumn import DataColumn
2from core.auxiliary.serializers.DataColumnSerializer import DataColumnSerializer
3from drf_spectacular.utils import OpenApiParameter, OpenApiTypes, extend_schema
4from rest_framework.decorators import api_view, renderer_classes, permission_classes
5from rest_framework.response import Response
6from rest_framework_csv.renderers import CSVRenderer
7from rest_framework import serializers
8from rest_framework.permissions import IsAuthenticated
9from authentication.permissions import HasExcelClientAccess, HasHumanUserAccess
10from authentication.token_exchange import exchange_token_for_excel_delegate
11import csv
12from django.http import HttpResponse
13from core.validation import api_view_validate
14from core.auxiliary.methods.export_scenario_data import export_scenario_data, collate
15from core.auxiliary.models.Scenario import Scenario
16from core.auxiliary.models.Flowsheet import Flowsheet
17import json
18from django.shortcuts import get_object_or_404
19from django.conf import settings
20from django.utils.html import escape
23def _escape_for_m_string(value: str) -> str:
24 return value.replace('"', '""')
26def transform_to_csv(data):
27 csv_data = []
28 for item in data:
29 for dataCell in item.get('dataCells'):
30 csv_data.append({'name': item.get('name'), 'value': dataCell.get('value')})
31 return csv_data
33class MSSRenderer (CSVRenderer):
34 header = ['name', 'value']
37class DownloadMSSResultsQuerySerializer(serializers.Serializer):
38 flowsheet = serializers.IntegerField(min_value=1)
39 scenario = serializers.IntegerField(min_value=1)
40 properties = serializers.ListField(
41 child=serializers.IntegerField(min_value=1),
42 required=False,
43 default=list,
44 )
46 @staticmethod
47 def _parse_properties_query_param(raw_properties: list[object]) -> list[int]:
48 if not raw_properties:
49 return []
51 if len(raw_properties) > 1:
52 values = raw_properties
53 else:
54 raw_properties_value = raw_properties[0]
55 if raw_properties_value == "": 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true
56 return []
58 try:
59 decoded = json.loads(raw_properties_value)
60 except (TypeError, json.JSONDecodeError):
61 values = [raw_properties_value]
62 else:
63 if isinstance(decoded, list): 63 ↛ 65line 63 didn't jump to line 65 because the condition on line 63 was always true
64 values = decoded
65 elif isinstance(decoded, (int, str)):
66 values = [decoded]
67 else:
68 raise serializers.ValidationError(
69 {
70 "properties": (
71 "Expected a JSON list of integers or repeated "
72 "'properties' query parameters."
73 )
74 }
75 )
77 parsed_properties: list[int] = []
78 for value in values:
79 try:
80 parsed_properties.append(int(value))
81 except (TypeError, ValueError):
82 raise serializers.ValidationError(
83 {"properties": "All property identifiers must be integers."}
84 )
86 return parsed_properties
88 def to_internal_value(self, data):
89 if hasattr(data, "getlist"): 89 ↛ 95line 89 didn't jump to line 95 because the condition on line 89 was always true
90 data = {
91 "flowsheet": data.get("flowsheet"),
92 "scenario": data.get("scenario"),
93 "properties": self._parse_properties_query_param(data.getlist("properties")),
94 }
95 elif "properties" in data:
96 raw_properties = data["properties"]
97 parsed_properties = self._parse_properties_query_param(
98 raw_properties if isinstance(raw_properties, list) else [raw_properties]
99 )
100 data = {
101 **data,
102 "properties": parsed_properties,
103 }
105 return super().to_internal_value(data)
108class DownloadExcelConnectionQuerySerializer(serializers.Serializer):
109 flowsheet = serializers.IntegerField(min_value=1)
110 scenario = serializers.IntegerField(min_value=1)
112@api_view_validate
113@extend_schema(parameters=[
114 OpenApiParameter(name="flowsheet", required=True, type=OpenApiTypes.INT),
115])
116@api_view(['GET'])
117@renderer_classes([MSSRenderer])
118def download_data(request):
119 """
120 This endpoint allows downloading of the input data for a multisteadystate simulation.
121 I am not sure if this has been updated to reflect the new Scenario model, or if
122 it is still used in the UI.
123 """
124 flowsheet_id = request.query_params.get("flowsheet")
125 if not flowsheet_id:
126 return Response(
127 {"error": "flowsheetId parameter is required."},
128 status=400
129 )
130 data_columns = DataColumn.objects.filter(flowsheet=flowsheet_id)
131 serialized_data = DataColumnSerializer(data_columns, many=True).data
133 csv_data = transform_to_csv(serialized_data)
135 response = HttpResponse(content_type='text/csv')
136 response['Content-Disposition'] = 'attachment; filename="export.csv"'
137 writer = csv.DictWriter(response, fieldnames=['name', 'value'])
138 writer.writeheader()
139 writer.writerows(csv_data)
141 return response
143@api_view_validate
144@extend_schema(parameters=[DownloadMSSResultsQuerySerializer])
145@api_view(['GET'])
146@permission_classes([IsAuthenticated, HasHumanUserAccess | HasExcelClientAccess])
147@renderer_classes([CSVRenderer])
148def download_mss_results(request):
149 """
150 This endpoint allows downloading of the results data for a multisteadystate simulation.
151 """
152 serializer = DownloadMSSResultsQuerySerializer(data=request.query_params)
153 serializer.is_valid(raise_exception=True)
154 validated_data = serializer.validated_data
156 flowsheet = get_object_or_404(Flowsheet, pk=validated_data["flowsheet"])
157 scenario = get_object_or_404(
158 Scenario,
159 pk=validated_data["scenario"],
160 flowsheet=flowsheet,
161 )
162 data = export_scenario_data(flowsheet, scenario, validated_data["properties"])
163 response = HttpResponse(content_type='text/csv')
164 response['Content-Disposition'] = 'attachment; filename="data.csv"'
165 writer = csv.DictWriter(response, fieldnames=data.keys())
166 writer.writeheader()
167 writer.writerows(collate(data))
169 return response
172@api_view_validate
173@extend_schema(parameters=[DownloadExcelConnectionQuerySerializer])
174@api_view(['GET'])
175@permission_classes([IsAuthenticated, HasHumanUserAccess])
176def download_excel_connection(request):
177 serializer = DownloadExcelConnectionQuerySerializer(data=request.query_params)
178 serializer.is_valid(raise_exception=True)
179 validated_data = serializer.validated_data
181 flowsheet = get_object_or_404(Flowsheet, pk=validated_data["flowsheet"])
182 scenario = get_object_or_404(
183 Scenario,
184 pk=validated_data["scenario"],
185 flowsheet=flowsheet,
186 )
188 subject_token = request.META.get(settings.REMOTE_USER_ACCESS_TOKEN_HEADER)
189 if not subject_token:
190 return Response({"error": "Missing upstream access token header."}, status=401)
192 if not settings.DEBUG: 192 ↛ 196line 192 didn't jump to line 196 because the condition on line 192 was always true
193 delegated_token = exchange_token_for_excel_delegate(subject_token)
194 else:
195 # In debug mode, we are not running keycloak. We can use the user's own token, since this is just for testing purposes.
196 delegated_token = subject_token
198 csv_url = (
199 f"{request.build_absolute_uri('/').rstrip('/')}/api/mss/download_results"
200 f"?flowsheet={flowsheet.pk}&scenario={scenario.pk}"
201 )
203 connection_name = escape(flowsheet.name or "Flowsheet")
204 safe_csv_url = _escape_for_m_string(csv_url)
205 safe_delegated_token = _escape_for_m_string(delegated_token)
207 m_formula = (
208 "let\n"
209 f" Source = Csv.Document(Web.Contents(\"{safe_csv_url}\", [Headers=[Authorization=\"Bearer {safe_delegated_token}\"]]),"
210 "[Delimiter=\",\", Encoding=1252, QuoteStyle=QuoteStyle.None]),\n"
211 " #\"Promoted Headers\" = Table.PromoteHeaders(Source, [PromoteAllScalars=true]),\n"
212 " #\"Changed Type\" = Table.TransformColumnTypes(#\"Promoted Headers\",List.Transform(\n"
213 " Table.ColumnNames(#\"Promoted Headers\"),\n"
214 " each {_, type number}\n"
215 " ))\n"
216 "in\n"
217 " #\"Changed Type\""
218 )
219 odc = f"""<html xmlns:o="urn:schemas-microsoft-com:office:office" xmlns="http://www.w3.org/TR/REC-html40">
220<head>
221<meta http-equiv=Content-Type content="text/x-ms-odc; charset=utf-8">
222<meta name=ProgId content=ODC.Database>
223<meta name=SourceType content=OLEDB>
224<title>Query - {connection_name}</title>
225<xml id=docprops><o:DocumentProperties xmlns:o="urn:schemas-microsoft-com:office:office" xmlns="http://www.w3.org/TR/REC-html40">
226 <o:Description>Connection to the '{connection_name}' query in the workbook.</o:Description>
227 <o:Name>Query - {connection_name}</o:Name>
228 </o:DocumentProperties>
229</xml><xml id=msodc><odc:OfficeDataConnection xmlns:odc="urn:schemas-microsoft-com:office:odc" xmlns="http://www.w3.org/TR/REC-html40">
230 <!-- odc:Type uses legacy OLEDB metadata while query execution is done by Power Query Web.Contents below. -->
231 <odc:PowerQueryConnection odc:Type="OLEDB">
232 <odc:ConnectionString>Provider=Microsoft.Mashup.OleDb.1;Data Source=$Workbook$;Location={connection_name};Extended Properties=""</odc:ConnectionString>
233 <odc:CommandType>SQL</odc:CommandType>
234 <odc:CommandText>SELECT * FROM [{connection_name}]</odc:CommandText>
235 </odc:PowerQueryConnection>
236 <odc:PowerQueryMashupData><Mashup xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://schemas.microsoft.com/DataMashup"><Client>EXCEL</Client><Version>2.147.1029.0</Version><MinVersion>2.21.0.0</MinVersion><Culture>en-GB</Culture><SafeCombine>true</SafeCombine><Items><Query Name="{connection_name}"><Formula><![CDATA[{m_formula}]]></Formula><IsParameterQuery xsi:nil="true" /><IsDirectQuery xsi:nil="true" /></Query></Items></Mashup></odc:PowerQueryMashupData>
237 </odc:OfficeDataConnection>
238</xml>
239</head>
240</html>"""
242 response = HttpResponse(odc, content_type="text/x-ms-odc; charset=utf-8")
243 response["Content-Disposition"] = 'attachment; filename="excel_connection.odc"'
244 return response