Coverage for backend/django/core/auxiliary/viewsets/ProjectViewSet.py: 93%
245 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from rest_framework import serializers, status, viewsets
2from rest_framework.exceptions import ValidationError
3from rest_framework.response import Response
4from rest_framework.decorators import action
6from django.db.models import F, QuerySet, Prefetch
7from django.db.models.functions import Lower
9from CoreRoot import settings
10from authentication.user.AccessTable import AccessTable
11from authentication.user.models import User
12from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType
13from core.auxiliary.models.Project import Project
14from core.auxiliary.models.ProjectFolder import ProjectFolder
15from core.auxiliary.models.Flowsheet import Flowsheet
16from core.auxiliary.serializers.ProjectSerializer import ProjectDetailSerializer, ProjectSerializer
17from core.auxiliary.serializers.FlowsheetCloneValidationSerializer import (
18 CloneRequestSerializer,
19 FlowsheetCloneValidationErrorSerializer,
20 ScenarioResultCopySummarySerializer,
21)
22from core.auxiliary.services.flowsheet_states import (
23 FlowsheetStateCloneError,
24 clone_validation_error_data,
25 copy_project_data,
26 summarize_scenario_results,
27)
28from core.managers import has_flowsheet_write_access, project_access_filter
29import core.auxiliary.enums.ViewType as ViewType
30from core.pagination import ClampedViewSetPagination
31from core.viewset import HistoricalRevisionWriteGuardMixin
33from drf_spectacular.utils import extend_schema, OpenApiParameter
34from drf_spectacular.types import OpenApiTypes
37class ProjectListPagination(ClampedViewSetPagination):
38 """Name the project-list pagination policy for generated API schemas."""
41class ProjectSharingSerializer(serializers.Serializer):
42 user_email = serializers.EmailField()
43 read_only = serializers.BooleanField(default=False)
46class RemoveSharedProjectUserSerializer(serializers.Serializer):
47 user_email = serializers.EmailField()
50class UpdateSharedProjectUserAccessSerializer(serializers.Serializer):
51 user_email = serializers.EmailField()
52 read_only = serializers.BooleanField()
55class SharedProjectUserSerializer(serializers.Serializer):
56 email = serializers.EmailField()
57 read_only = serializers.BooleanField()
60class ListSharedProjectUsersSerializer(serializers.Serializer):
61 users = SharedProjectUserSerializer(many=True)
64class ProjectViewSet(HistoricalRevisionWriteGuardMixin, viewsets.ModelViewSet):
65 serializer_class = ProjectSerializer
66 lookup_value_regex = r"\d+"
67 pagination_class = ProjectListPagination
69 PROJECT_LIST_ORDERING = {
70 "recently_added": (F("created_at").desc(nulls_last=True), "-id"),
71 "recently_binned": (F("binned_at").desc(nulls_last=True), "-id"),
72 "recently_edited": (
73 F("active_flowsheet__savedDate").desc(nulls_last=True),
74 "-id",
75 ),
76 "name": (Lower("name").asc(), "id"),
77 }
79 def get_serializer_class(self):
80 """
81 Use lightweight serializer for listing projects.
82 Detailed serializer for retreiving a single project so it inlcudes all flowsheet configurations (to display in the frontend sidebar).
83 """
84 if self.action == "retrieve":
85 return ProjectDetailSerializer
86 return ProjectSerializer
88 def _base_queryset(self) -> QuerySet[Project]:
89 return Project.objects.select_related("owner","active_flowsheet")
91 def _detail_queryset(self, user) -> QuerySet[Project]:
92 return (
93 self._all_visible_queryset(user).prefetch_related(
94 Prefetch(
95 "flowsheets",
96 queryset=(
97 Flowsheet.objects
98 .select_related("owner")
99 .order_by("created_at", "id")
100 ),
101 to_attr="prefetched_configurations"
102 )
103 )
104 )
106 def _owned_queryset(self, user) -> QuerySet[Project]:
107 """Projects owned by the user"""
108 return self._base_queryset().filter(owner=user)
110 def _shared_queryset(self, user) -> QuerySet[Project]:
111 """Projects shared with the user"""
112 return (
113 self._base_queryset()
114 .filter(access_list__user=user,)
115 .exclude(owner=user)
116 .distinct()
117 )
119 def _all_visible_queryset(self, user) -> QuerySet[Project]:
120 """All projects visible to the user, including owned and shared"""
121 return (
122 self._base_queryset()
123 .filter(project_access_filter(user, project_path=None))
124 .distinct()
125 )
127 def _project_list_queryset(self, queryset: QuerySet[Project]) -> QuerySet[Project]:
128 """Return only projects whose active flowsheet is a regular editable flowsheet."""
130 return queryset.filter(
131 active_flowsheet__flowsheet_template_type=FlowsheetTemplateType.NotTemplate,
132 )
134 def get_queryset(self) -> QuerySet[Project]:
135 return self._owned_queryset(self.request.user)
137 def _get_owned_project(self, user, pk):
138 return self._owned_queryset(user).get(pk=pk)
140 def _validate_project_share(self, owner, project, sharee):
141 if sharee.email == owner.email:
142 raise ValueError("You cannot share a project with yourself.")
144 if sharee.email == settings.PLATFORM_TEST_EMAIL and not owner.is_tester:
145 raise PermissionError("Only testers can share with the platform test account.")
147 if AccessTable.objects.filter(user=sharee, project=project).exists():
148 raise ValueError("This user already has access to the project.")
150 def _shared_project_users(self, project):
151 return (
152 AccessTable.objects
153 .filter(project=project)
154 .exclude(user=project.owner)
155 .values("user__email", "read_only")
156 .order_by("user__email")
157 )
159 @extend_schema(
160 parameters=[
161 OpenApiParameter(name="type", required=True, type=OpenApiTypes.STR),
162 OpenApiParameter(
163 name="folder",
164 required=False,
165 type=OpenApiTypes.STR,
166 description="Owned-project folder ID, or 'root' for unfiled projects.",
167 ),
168 OpenApiParameter(name="is_starred", required=False, type=OpenApiTypes.BOOL),
169 OpenApiParameter(name="is_binned", required=False, type=OpenApiTypes.BOOL),
170 OpenApiParameter(
171 name="ordering",
172 required=False,
173 type=OpenApiTypes.STR,
174 enum=list(PROJECT_LIST_ORDERING),
175 ),
176 OpenApiParameter(name="page", required=False, type=OpenApiTypes.INT),
177 ]
178 )
179 def list(self, request, *args, **kwargs):
180 """Return one filtered and deterministically ordered project page."""
182 view_type = request.query_params.get("type", ViewType.OWNED)
184 if view_type == ViewType.SHARED:
185 queryset = self._shared_queryset(request.user)
186 elif view_type == ViewType.ALL:
187 queryset = self._all_visible_queryset(request.user)
188 else:
189 queryset = self._owned_queryset(request.user)
191 queryset = self._project_list_queryset(queryset)
192 queryset = self._filter_project_list(queryset, view_type=view_type)
193 ordering = request.query_params.get("ordering", "recently_edited")
194 try:
195 ordering_fields = self.PROJECT_LIST_ORDERING[ordering]
196 except KeyError as exc:
197 raise ValidationError({"ordering": "Unknown project ordering."}) from exc
198 queryset = queryset.order_by(*ordering_fields)
200 page = self.paginate_queryset(queryset)
201 serializer = self.get_serializer(page, many=True)
202 return self.get_paginated_response(serializer.data)
204 def _filter_project_list(
205 self,
206 queryset: QuerySet[Project],
207 *,
208 view_type: str,
209 ) -> QuerySet[Project]:
210 """Apply explicit list filters before pagination."""
212 folder = self.request.query_params.get("folder")
213 if folder is not None:
214 if view_type != ViewType.OWNED: 214 ↛ 215line 214 didn't jump to line 215 because the condition on line 214 was never true
215 raise ValidationError(
216 {"folder": "Folder filtering is only available for owned projects."}
217 )
218 if folder == "root":
219 queryset = queryset.filter(folder__isnull=True)
220 else:
221 try:
222 folder_id = int(folder)
223 except ValueError as exc:
224 raise ValidationError({"folder": "Use a folder ID or 'root'."}) from exc
225 if not ProjectFolder.objects.filter(
226 pk=folder_id,
227 owner=self.request.user,
228 is_binned=False,
229 ).exists():
230 raise ValidationError({"folder": "Project folder not found."})
231 queryset = queryset.filter(folder_id=folder_id)
233 for parameter in ("is_starred", "is_binned"):
234 raw_value = self.request.query_params.get(parameter)
235 if raw_value is None:
236 continue
237 normalized_value = raw_value.casefold()
238 if normalized_value not in {"true", "false"}: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 raise ValidationError({parameter: "Use true or false."})
240 queryset = queryset.filter(**{parameter: normalized_value == "true"})
241 return queryset
243 def retrieve(self, request, *args, **kwargs):
244 try:
245 instance = self._detail_queryset(request.user).get(pk=kwargs["pk"])
246 except Project.DoesNotExist:
247 return Response({"error": "Project not found."}, status=status.HTTP_404_NOT_FOUND)
249 serializer = self.get_serializer(instance)
250 return Response(serializer.data)
252 def create(self, request, *args, **kwargs):
253 serializer = self.get_serializer(data=request.data, context={"request": request})
254 serializer.is_valid(raise_exception=True)
255 serializer.save()
257 return Response(serializer.data, status=status.HTTP_201_CREATED)
259 def partial_update(self, request, *args, **kwargs):
260 """
261 Owners use the standard serializer update path.
262 Non-owners are routed to a restricted shared-project path so menu-bar rename can work without changing owner-only main-page project actions.
263 """
264 pk = kwargs["pk"]
266 try:
267 project = self._owned_queryset(request.user).get(pk=pk)
268 except Project.DoesNotExist:
269 return self._partial_update_shared_project(request, pk)
271 serializer = self.get_serializer(project, data=request.data, partial=True)
272 serializer.is_valid(raise_exception=True)
273 serializer.save()
274 return Response(serializer.data)
276 def _partial_update_shared_project(self, request, pk):
277 """
278 Shared users with write access may only patch the project name. This supports menu-bar renaming while keeping fields like starred/binned owner-controlled.
279 """
280 try:
281 project = self._all_visible_queryset(request.user).get(pk=pk)
282 except Project.DoesNotExist:
283 return Response({"error": "Project not found."}, status=status.HTTP_404_NOT_FOUND)
285 access = AccessTable.objects.filter(user=request.user, project=project).first()
286 if access is None or access.read_only:
287 return Response(
288 {"error": "You do not have permission to update this project."},
289 status=status.HTTP_403_FORBIDDEN,
290 )
292 if set(request.data.keys()) != {"name"}:
293 return Response(
294 {"error": "Shared users with write access can only update the project name."},
295 status=status.HTTP_403_FORBIDDEN,
296 )
298 serializer = self.get_serializer(project, data={"name": request.data["name"]}, partial=True)
299 serializer.is_valid(raise_exception=True)
300 serializer.save()
301 return Response(serializer.data)
303 @extend_schema(request=ProjectSharingSerializer, responses=None)
304 @action(detail=True, methods=["post"], url_path="share")
305 def share_project(self, request, pk=None):
306 serializer = ProjectSharingSerializer(data=request.data)
307 serializer.is_valid(raise_exception=True)
308 user_email = serializer.validated_data["user_email"]
309 read_only = serializer.validated_data["read_only"]
311 try:
312 project = self._get_owned_project(request.user, pk)
313 sharee = User.objects.get(email=user_email)
315 self._validate_project_share(request.user, project, sharee)
316 AccessTable.objects.create(user=sharee, project=project, read_only=read_only)
317 return Response({"message": "Project shared successfully"}, status=status.HTTP_200_OK)
318 except ValueError as exc:
319 return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
320 except PermissionError as exc:
321 return Response({"error": str(exc)}, status=status.HTTP_403_FORBIDDEN)
322 except (Project.DoesNotExist, User.DoesNotExist):
323 return Response({"error": "Failed to share project"}, status=status.HTTP_404_NOT_FOUND)
325 @extend_schema(responses=ListSharedProjectUsersSerializer)
326 @action(detail=True, methods=["get"], url_path="shared-users")
327 def list_shared_users(self, request, pk=None):
328 try:
329 project = self._get_owned_project(request.user, pk)
330 except Project.DoesNotExist:
331 return Response({"error": "Failed to retrieve list of user"}, status=status.HTTP_404_NOT_FOUND)
333 serializer = ListSharedProjectUsersSerializer({
334 "users": [
335 {"email": row["user__email"], "read_only": row["read_only"]}
336 for row in self._shared_project_users(project)
337 ]
338 })
339 return Response(serializer.data, status=status.HTTP_200_OK)
341 @extend_schema(request=RemoveSharedProjectUserSerializer, responses=None)
342 @action(detail=True, methods=["post"], url_path="remove-user")
343 def remove_user(self, request, pk=None):
344 serializer = RemoveSharedProjectUserSerializer(data=request.data)
345 serializer.is_valid(raise_exception=True)
346 user_email = serializer.validated_data["user_email"]
348 try:
349 project = self._get_owned_project(request.user, pk)
350 user = User.objects.get(email=user_email)
352 if user.id == project.owner_id: 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true
353 raise ValueError("Cannot remove the owner of the project.")
355 AccessTable.objects.get(user=user, project=project).delete()
356 return Response({"message": "User removed successfully"}, status=status.HTTP_200_OK)
357 except ValueError as exc:
358 return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
359 except (Project.DoesNotExist, User.DoesNotExist, AccessTable.DoesNotExist):
360 return Response({"error": "Failed to remove user"}, status=status.HTTP_404_NOT_FOUND)
362 @extend_schema(request=UpdateSharedProjectUserAccessSerializer, responses=None)
363 @action(detail=True, methods=["post"], url_path="update-shared-user-access")
364 def update_shared_user_access(self, request, pk=None):
365 serializer = UpdateSharedProjectUserAccessSerializer(data=request.data)
366 serializer.is_valid(raise_exception=True)
367 user_email = serializer.validated_data["user_email"]
368 read_only = serializer.validated_data["read_only"]
370 try:
371 project = self._get_owned_project(request.user, pk)
372 user = User.objects.get(email=user_email)
374 if user.id == project.owner_id:
375 raise ValueError("Cannot modify the owner's access.")
377 access_row = AccessTable.objects.get(user=user, project=project)
378 access_row.read_only = read_only
379 access_row.save(update_fields=["read_only"])
381 return Response({"message": "Shared user access updated successfully"}, status=status.HTTP_200_OK)
382 except ValueError as exc:
383 return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
384 except (Project.DoesNotExist, User.DoesNotExist, AccessTable.DoesNotExist):
385 return Response({"error": "Failed to update shared user access"}, status=status.HTTP_404_NOT_FOUND)
387 @extend_schema(
388 responses={200: ScenarioResultCopySummarySerializer},
389 description="Count scenario results available for a full project duplicate.",
390 )
391 @action(detail=True, methods=["get"], url_path="copy-preview")
392 def copy_preview(self, request, pk=None):
393 """Return counts across the non-template flowsheets copied with a project."""
395 try:
396 source_project = self._all_visible_queryset(request.user).get(pk=pk)
397 except Project.DoesNotExist:
398 return Response(
399 {"error": "Project not found."},
400 status=status.HTTP_404_NOT_FOUND,
401 )
402 state_ids = source_project.flowsheets.filter(
403 flowsheet_template_type=FlowsheetTemplateType.NotTemplate,
404 ).values_list("current_state_id", flat=True)
405 summary = summarize_scenario_results(flowsheet_state_ids=state_ids)
406 return Response(ScenarioResultCopySummarySerializer(summary).data)
408 @extend_schema(
409 request=CloneRequestSerializer,
410 responses={
411 201: ProjectSerializer,
412 422: FlowsheetCloneValidationErrorSerializer,
413 },
414 )
415 @action(detail=True, methods=["post"], url_path="copy")
416 def copy(self, request, pk=None):
417 input_serializer = CloneRequestSerializer(data=request.data)
418 input_serializer.is_valid(raise_exception=True)
419 repair_legacy_formula_units = input_serializer.validated_data[
420 "repair_legacy_formula_units"
421 ]
422 include_scenario_results = input_serializer.validated_data[
423 "include_scenario_results"
424 ]
425 try:
426 source_project = self._all_visible_queryset(request.user).get(pk=pk)
427 except Project.DoesNotExist:
428 return Response(
429 {"error: Project not found."},
430 status=status.HTTP_404_NOT_FOUND,
431 )
433 if repair_legacy_formula_units:
434 repairable_flowsheet_ids = source_project.flowsheets.filter(
435 flowsheet_template_type=FlowsheetTemplateType.NotTemplate,
436 ).values_list("pk", flat=True)
437 if any( 437 ↛ 441line 437 didn't jump to line 441 because the condition on line 437 was never true
438 not has_flowsheet_write_access(request.user, flowsheet_id)
439 for flowsheet_id in repairable_flowsheet_ids
440 ):
441 return Response(
442 {"error": "You do not have write access to repair this project"},
443 status=status.HTTP_403_FORBIDDEN,
444 )
446 try:
447 copied_project = copy_project_data(
448 source_project=source_project,
449 user=request.user,
450 repair_legacy_formula_units=repair_legacy_formula_units,
451 include_scenario_results=include_scenario_results,
452 )
453 except FlowsheetStateCloneError as exc:
454 if not exc.is_user_repairable: 454 ↛ 455line 454 didn't jump to line 455 because the condition on line 454 was never true
455 raise
456 return Response(
457 clone_validation_error_data(exc),
458 status=status.HTTP_422_UNPROCESSABLE_ENTITY,
459 )
461 return Response(
462 self.get_serializer(copied_project).data,
463 status=status.HTTP_201_CREATED,
464 )