Coverage for backend/django/core/auxiliary/serializers/ProjectSerializer.py: 86%
121 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.core.exceptions import ValidationError
2from django.db import transaction
3from django.db.models import Q
4from drf_spectacular.types import OpenApiTypes
5from drf_spectacular.utils import extend_schema_field
6from rest_framework import serializers
8from core.auxiliary.models.Project import Project
9from core.auxiliary.models.ProjectFolder import ProjectFolder
10from core.auxiliary.models.Flowsheet import Flowsheet
11from core.auxiliary.serializers.FlowsheetSerializer import OwnerSerializer
12from core.auxiliary.services.project_folders import (
13 PROJECT_FOLDER_REGULAR_PROJECT_ERROR,
14 ProjectFolderHierarchyError,
15 ProjectFolderNotFoundError,
16 ProjectUpdateError,
17 apply_project_update,
18 lock_active_project_folder,
19 project_has_regular_active_flowsheet,
20)
23class ProjectSerializer(serializers.ModelSerializer):
24 """
25 Serializer for Project list.
26 Lightweight: does not return the full list of flowsheet configurations in the project.
27 """
29 owner = OwnerSerializer(read_only=True)
30 savedDate = serializers.SerializerMethodField()
31 active_flowsheet_rootGrouping = serializers.SerializerMethodField()
33 class Meta:
34 model = Project
35 fields = "__all__"
36 read_only_fields = [
37 "owner",
38 "active_flowsheet",
39 "binned_at",
40 "savedDate",
41 "active_flowsheet_rootGrouping",
42 ]
44 def __init__(self, *args, **kwargs):
45 super().__init__(*args, **kwargs)
46 request = self.context.get("request")
47 user = getattr(request, "user", None)
48 if getattr(user, "is_authenticated", False):
49 allowed_folders = Q(owner=user, is_binned=False)
50 # A restore payload may round-trip the project's retained binned
51 # folder ID. Object validation below admits that row only when the
52 # same request explicitly restores this project.
53 if (
54 isinstance(self.instance, Project)
55 and self.instance.owner_id == user.id
56 and self.instance.folder_id is not None
57 ):
58 allowed_folders |= Q(
59 owner=user,
60 pk=self.instance.folder_id,
61 )
62 self.fields["folder"].queryset = ProjectFolder.objects.filter(
63 allowed_folders
64 )
66 def get_savedDate(self, obj: Project):
67 if obj.active_flowsheet is None: 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true
68 return None
69 return obj.active_flowsheet.savedDate
71 def get_active_flowsheet_rootGrouping(self, obj: Project):
72 if obj.active_flowsheet is None: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true
73 return None
74 if obj.active_flowsheet.current_state_id is None: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true
75 return None
76 return obj.active_flowsheet.current_state.root_grouping_id
78 @transaction.atomic
79 def create(self, validated_data):
80 """Create a project after locking and revalidating its destination folder."""
82 context = self.context
83 request = context["request"]
84 owner = request.user
86 # The default flowsheet is assigned after the project shell is created.
87 validated_data.pop("active_flowsheet", None)
88 requests_binning = validated_data.pop("is_binned", False)
90 folder = validated_data.get("folder")
91 if folder is not None:
92 try:
93 validated_data["folder"] = lock_active_project_folder(
94 owner_id=owner.pk,
95 folder_id=folder.pk,
96 )
97 except ProjectFolderNotFoundError as exc:
98 raise serializers.ValidationError({"folder": str(exc)}) from exc
100 project = Project.create(**validated_data, owner=owner)
101 if not requests_binning:
102 return project
104 # Creation first establishes the regular active flowsheet required by
105 # the canonical bin transition. The surrounding transaction rolls the
106 # complete create back if that transition cannot be applied.
107 try:
108 return apply_project_update(
109 stale_project=project,
110 changes={"is_binned": True},
111 )
112 except ProjectUpdateError as exc:
113 raise serializers.ValidationError({exc.field: str(exc)}) from exc
115 def validate_folder(self, folder):
116 """Restrict project membership changes to active owner folders."""
118 request = self.context.get("request")
119 user = getattr(request, "user", None)
120 if user is None: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true
121 if folder is not None:
122 raise serializers.ValidationError(
123 "Project folder ownership could not be verified."
124 )
125 return None
127 if self.instance is not None and self.instance.owner_id != user.id: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 raise serializers.ValidationError("Only the project owner can move this project.")
130 if self.instance is not None and not project_has_regular_active_flowsheet(
131 self.instance
132 ):
133 raise serializers.ValidationError(PROJECT_FOLDER_REGULAR_PROJECT_ERROR)
135 if folder is not None:
136 restoring_current_folder = bool(
137 self.instance is not None
138 and self.instance.is_binned
139 and self.initial_data.get("is_binned") is False
140 and folder.pk == self.instance.folder_id
141 )
142 if ( 142 ↛ 146line 142 didn't jump to line 146 because the condition on line 142 was never true
143 folder.owner_id != user.id
144 or (folder.is_binned and not restoring_current_folder)
145 ):
146 raise serializers.ValidationError("Project folder not found.")
147 return folder
149 def update(self, instance, validated_data):
150 """Delegate project membership and lifecycle transitions to the domain service."""
152 try:
153 return apply_project_update(
154 stale_project=instance,
155 changes=validated_data,
156 )
157 except (ProjectFolderNotFoundError, ProjectFolderHierarchyError) as exc:
158 raise serializers.ValidationError({"folder": str(exc)}) from exc
159 except ProjectUpdateError as exc:
160 raise serializers.ValidationError({exc.field: str(exc)}) from exc
162 def to_representation(self, instance):
163 """Hide owner-only folder organization from shared-project responses."""
165 representation = super().to_representation(instance)
166 request = self.context.get("request")
167 request_user_id = getattr(getattr(request, "user", None), "id", None)
168 if instance.owner_id != request_user_id:
169 representation["folder"] = None
170 return representation
172 def validate(self, attrs):
173 if self.instance is None:
174 if "binned_at" in self.initial_data:
175 raise serializers.ValidationError(
176 {
177 "binned_at": (
178 "Project bin timestamps cannot be set during creation."
179 )
180 }
181 )
183 if self.instance is None and attrs.get("active_flowsheet") is not None: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true
184 raise serializers.ValidationError(
185 {
186 "active_flowsheet": (
187 "Default flowsheet is set automatically on project creation."
188 )
189 }
190 )
192 if self.instance is None:
193 request = self.context.get("request")
194 project = Project(owner=getattr(request, "user", None), **attrs)
195 else:
196 # DRF calls object validation before ``update``. Validate a detached
197 # candidate so lock planning still sees the original instance
198 # snapshot (especially its current folder and bin state).
199 project = Project(
200 **{
201 field.attname: getattr(self.instance, field.attname)
202 for field in Project._meta.concrete_fields
203 }
204 )
205 for field, value in attrs.items():
206 setattr(project, field, value)
208 # Ensure that the default flowsheet belongs to the project
209 try:
210 project.clean()
211 except ValidationError as e:
212 raise serializers.ValidationError(e.message_dict)
214 return attrs
217class ProjectConfigSummarySerializer(serializers.ModelSerializer):
218 """
219 Defines the shape of each flowsheet in the list of configurations.
220 """
221 is_active = serializers.SerializerMethodField()
223 class Meta:
224 model = Flowsheet
225 fields = [
226 "id",
227 "name",
228 "savedDate",
229 "created_at",
230 "flowsheet_template_type",
231 "is_active",
232 ]
234 @extend_schema_field(OpenApiTypes.BOOL)
235 def get_is_active(self, obj: Flowsheet):
236 active_flowsheet_id = self.context.get("active_flowsheet_id")
237 return obj.id == active_flowsheet_id
239class ProjectDetailSerializer(ProjectSerializer):
240 """
241 Serializer for retrieving a single project. This includes the full list of flowsheet configurations in that project.
242 """
243 configurations = serializers.SerializerMethodField()
245 @extend_schema_field(ProjectConfigSummarySerializer(many=True))
246 def get_configurations(self, obj: Project):
247 configs = obj.get_configurations()
248 serializer = (
249 ProjectConfigSummarySerializer(
250 configs,
251 many=True,
252 context={
253 **self.context,
254 "active_flowsheet_id": obj.active_flowsheet_id,
255 },
256 )
257 )
258 return serializer.data