435 lines
17 KiB
Python
435 lines
17 KiB
Python
# Copyright (c) 2023-present Plane Software, Inc. and contributors
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# See the LICENSE file for details.
|
|
|
|
from rest_framework import serializers
|
|
|
|
from .base import BaseSerializer
|
|
from .issue import IssueSerializer
|
|
from .project import ProjectLiteSerializer
|
|
from .state import StateLiteSerializer
|
|
from .user import UserLiteSerializer
|
|
from plane.app.serializers.issue import LabelSerializer
|
|
from plane.db.models import FileAsset, IntakeIssue, Issue, IssueActivity, IssueComment, Label, Notification, Project
|
|
|
|
|
|
class ExternalContourIssuePayloadSerializer(serializers.Serializer):
|
|
name = serializers.CharField(max_length=255)
|
|
description_html = serializers.CharField(required=False, allow_blank=True, allow_null=True)
|
|
priority = serializers.ChoiceField(choices=Issue.PRIORITY_CHOICES, default="none", required=False)
|
|
assignee_ids = serializers.ListField(child=serializers.UUIDField(), required=False)
|
|
label_ids = serializers.ListField(child=serializers.UUIDField(), required=False)
|
|
target_date = serializers.DateField(required=False, allow_null=True)
|
|
|
|
|
|
class ExternalContourRequestCreateSerializer(serializers.Serializer):
|
|
target_project_id = serializers.UUIDField()
|
|
issue = ExternalContourIssuePayloadSerializer()
|
|
|
|
|
|
class ExternalContourRequestDecisionSerializer(serializers.Serializer):
|
|
action = serializers.ChoiceField(choices=["accept", "decline"])
|
|
comment = serializers.CharField(required=False, allow_blank=True)
|
|
|
|
def validate(self, data):
|
|
if data.get("action") == "decline" and not (data.get("comment") or "").strip():
|
|
raise serializers.ValidationError({"comment": "Decline reason is required"})
|
|
return data
|
|
|
|
|
|
class ExternalContourRequestReplySerializer(serializers.Serializer):
|
|
comment = serializers.CharField()
|
|
|
|
|
|
class ExternalContourRequestUpdateSerializer(serializers.Serializer):
|
|
name = serializers.CharField(max_length=255, required=False)
|
|
description_html = serializers.CharField(required=False, allow_blank=True, allow_null=True)
|
|
priority = serializers.ChoiceField(choices=Issue.PRIORITY_CHOICES, required=False)
|
|
assignee_ids = serializers.ListField(child=serializers.UUIDField(), required=False)
|
|
label_ids = serializers.ListField(child=serializers.UUIDField(), required=False)
|
|
state_id = serializers.UUIDField(required=False)
|
|
target_date = serializers.DateField(required=False, allow_null=True)
|
|
|
|
def validate(self, data):
|
|
if not data:
|
|
raise serializers.ValidationError("At least one field must be provided")
|
|
|
|
if "name" in data:
|
|
data["name"] = data["name"].strip()
|
|
if not data["name"]:
|
|
raise serializers.ValidationError({"name": "Title is required"})
|
|
|
|
return data
|
|
|
|
|
|
class ExternalContourTargetProjectSerializer(BaseSerializer):
|
|
inbox_view = serializers.BooleanField(read_only=True, source="intake_view")
|
|
|
|
class Meta:
|
|
model = Project
|
|
fields = ["id", "identifier", "name", "logo_props", "inbox_view"]
|
|
read_only_fields = fields
|
|
|
|
|
|
class ExternalContourTargetOptionsSerializer(serializers.Serializer):
|
|
project = ExternalContourTargetProjectSerializer(read_only=True)
|
|
member_ids = serializers.ListField(child=serializers.UUIDField(), read_only=True)
|
|
states = StateLiteSerializer(many=True, read_only=True)
|
|
labels = LabelSerializer(many=True, read_only=True)
|
|
|
|
|
|
class ExternalContourIssueSerializer(BaseSerializer):
|
|
assignee_ids = serializers.SerializerMethodField()
|
|
assignee_details = serializers.SerializerMethodField()
|
|
created_by_detail = UserLiteSerializer(source="created_by", read_only=True)
|
|
label_details = serializers.SerializerMethodField()
|
|
label_ids = serializers.SerializerMethodField()
|
|
project_detail = ProjectLiteSerializer(source="project", read_only=True)
|
|
state_detail = StateLiteSerializer(source="state", read_only=True)
|
|
|
|
class Meta:
|
|
model = Issue
|
|
fields = [
|
|
"id",
|
|
"name",
|
|
"description_html",
|
|
"priority",
|
|
"sequence_id",
|
|
"project_id",
|
|
"created_at",
|
|
"updated_at",
|
|
"created_by",
|
|
"state_id",
|
|
"target_date",
|
|
"label_ids",
|
|
"label_details",
|
|
"assignee_ids",
|
|
"assignee_details",
|
|
"state_detail",
|
|
"project_detail",
|
|
"created_by_detail",
|
|
]
|
|
read_only_fields = fields
|
|
|
|
def get_assignee_ids(self, obj):
|
|
return [assignee.assignee_id for assignee in obj.issue_assignee.all()]
|
|
|
|
def get_assignee_details(self, obj):
|
|
return UserLiteSerializer([assignee.assignee for assignee in obj.issue_assignee.all()], many=True).data
|
|
|
|
def get_label_ids(self, obj):
|
|
return [label.label_id for label in obj.label_issue.all()]
|
|
|
|
def get_label_details(self, obj):
|
|
return [
|
|
{"id": str(label_bridge.label.id), "name": label_bridge.label.name, "color": label_bridge.label.color}
|
|
for label_bridge in obj.label_issue.all()
|
|
]
|
|
|
|
|
|
class ExternalContourMirroredAttachmentSerializer(BaseSerializer):
|
|
uploaded_by = serializers.SerializerMethodField()
|
|
download_url = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = FileAsset
|
|
fields = ["id", "attributes", "asset_url", "download_url", "updated_at", "uploaded_by"]
|
|
read_only_fields = fields
|
|
|
|
def get_uploaded_by(self, obj):
|
|
user = obj.updated_by or obj.created_by
|
|
return getattr(user, "display_name", None)
|
|
|
|
def get_download_url(self, obj):
|
|
workspace_slug = self.context.get("workspace_slug")
|
|
source_project_id = self.context.get("source_project_id")
|
|
request_id = self.context.get("request_id")
|
|
if not workspace_slug or not source_project_id or not request_id:
|
|
return None
|
|
return (
|
|
f"/api/workspaces/{workspace_slug}/projects/{source_project_id}/external-contours/"
|
|
f"{request_id}/attachments/{obj.id}/"
|
|
)
|
|
|
|
|
|
class ExternalContourMirroredCommentSerializer(BaseSerializer):
|
|
actor_detail = UserLiteSerializer(read_only=True, source="actor")
|
|
|
|
class Meta:
|
|
model = IssueComment
|
|
fields = ["id", "comment_html", "created_at", "updated_at", "edited_at", "parent_id", "actor_detail"]
|
|
read_only_fields = fields
|
|
|
|
|
|
class ExternalContourMirroredActivitySerializer(BaseSerializer):
|
|
actor_detail = UserLiteSerializer(read_only=True, source="actor")
|
|
|
|
class Meta:
|
|
model = IssueActivity
|
|
fields = ["id", "verb", "field", "old_value", "new_value", "comment", "created_at", "actor_detail"]
|
|
read_only_fields = fields
|
|
|
|
|
|
class ExternalContourRequestSerializer(BaseSerializer):
|
|
has_unread_updates = serializers.SerializerMethodField()
|
|
issue = ExternalContourIssueSerializer(read_only=True)
|
|
mirrored_activity = serializers.SerializerMethodField()
|
|
mirrored_attachments = serializers.SerializerMethodField()
|
|
mirrored_comments = serializers.SerializerMethodField()
|
|
source_project_id = serializers.SerializerMethodField()
|
|
source_project_name = serializers.SerializerMethodField()
|
|
source_decision = serializers.SerializerMethodField()
|
|
source_decision_at = serializers.SerializerMethodField()
|
|
source_decision_by_name = serializers.SerializerMethodField()
|
|
target_project_id = serializers.SerializerMethodField()
|
|
target_project_name = serializers.SerializerMethodField()
|
|
requested_by_id = serializers.SerializerMethodField()
|
|
requested_by_name = serializers.SerializerMethodField()
|
|
requested_at = serializers.SerializerMethodField()
|
|
status = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = IntakeIssue
|
|
fields = [
|
|
"id",
|
|
"created_at",
|
|
"updated_at",
|
|
"created_by",
|
|
"has_unread_updates",
|
|
"issue",
|
|
"mirrored_activity",
|
|
"mirrored_attachments",
|
|
"mirrored_comments",
|
|
"source_project_id",
|
|
"source_project_name",
|
|
"source_decision",
|
|
"source_decision_at",
|
|
"source_decision_by_name",
|
|
"target_project_id",
|
|
"target_project_name",
|
|
"requested_by_id",
|
|
"requested_by_name",
|
|
"requested_at",
|
|
"status",
|
|
]
|
|
read_only_fields = fields
|
|
|
|
def get_source_project_id(self, obj):
|
|
return obj.extra.get("source_project_id")
|
|
|
|
def get_has_unread_updates(self, obj):
|
|
annotated_value = getattr(obj, "has_unread_updates_annotated", None)
|
|
if annotated_value is not None:
|
|
return annotated_value
|
|
|
|
request = self.context.get("request")
|
|
user = getattr(request, "user", None)
|
|
user_id = getattr(user, "id", None)
|
|
if not user_id:
|
|
return False
|
|
|
|
return Notification.objects.filter(
|
|
receiver_id=user_id,
|
|
sender__startswith="in_app:external_contours:",
|
|
read_at__isnull=True,
|
|
data__issue__id=str(obj.id),
|
|
).exists()
|
|
|
|
def get_mirrored_activity(self, obj):
|
|
if not self.context.get("include_mirror_data") or not obj.issue_id:
|
|
return []
|
|
|
|
activity = (
|
|
IssueActivity.objects.filter(issue_id=obj.issue_id)
|
|
.exclude(field__in=["comment", "vote", "reaction", "draft"])
|
|
.select_related("actor")
|
|
.order_by("-created_at")[:50]
|
|
)
|
|
return ExternalContourMirroredActivitySerializer(activity, many=True).data
|
|
|
|
def get_mirrored_attachments(self, obj):
|
|
if not self.context.get("include_mirror_data") or not obj.issue_id:
|
|
return []
|
|
|
|
attachments = (
|
|
FileAsset.objects.filter(
|
|
issue_id=obj.issue_id,
|
|
entity_type=FileAsset.EntityTypeContext.ISSUE_ATTACHMENT,
|
|
is_uploaded=True,
|
|
is_deleted=False,
|
|
)
|
|
.select_related("created_by", "updated_by")
|
|
.order_by("-updated_at")
|
|
)
|
|
return ExternalContourMirroredAttachmentSerializer(
|
|
attachments,
|
|
many=True,
|
|
context={
|
|
"workspace_slug": self.context.get("workspace_slug"),
|
|
"source_project_id": self.context.get("source_project_id"),
|
|
"request_id": str(obj.id),
|
|
},
|
|
).data
|
|
|
|
def get_mirrored_comments(self, obj):
|
|
if not self.context.get("include_mirror_data") or not obj.issue_id:
|
|
return []
|
|
|
|
comments = (
|
|
IssueComment.objects.filter(issue_id=obj.issue_id)
|
|
.select_related("actor")
|
|
.order_by("created_at")
|
|
)
|
|
return ExternalContourMirroredCommentSerializer(comments, many=True).data
|
|
|
|
def get_source_project_name(self, obj):
|
|
source_project_id = obj.extra.get("source_project_id")
|
|
if source_project_id:
|
|
live_name = Project.objects.filter(pk=source_project_id).values_list("name", flat=True).first()
|
|
if live_name:
|
|
return live_name
|
|
return obj.extra.get("source_project_name")
|
|
|
|
def get_source_decision(self, obj):
|
|
return obj.extra.get("source_decision")
|
|
|
|
def get_source_decision_at(self, obj):
|
|
return obj.extra.get("source_decision_at")
|
|
|
|
def get_source_decision_by_name(self, obj):
|
|
return obj.extra.get("source_decision_by_name")
|
|
|
|
def get_target_project_id(self, obj):
|
|
target_project_id = obj.extra.get("target_project_id")
|
|
if target_project_id:
|
|
return target_project_id
|
|
if obj.issue and obj.issue.project_id:
|
|
return str(obj.issue.project_id)
|
|
return None
|
|
|
|
def get_target_project_name(self, obj):
|
|
target_project_name = obj.extra.get("target_project_name")
|
|
if target_project_name:
|
|
return target_project_name
|
|
if obj.issue and obj.issue.project:
|
|
return obj.issue.project.name
|
|
return None
|
|
|
|
def get_requested_by_id(self, obj):
|
|
requested_by_id = obj.extra.get("requested_by_id")
|
|
if requested_by_id:
|
|
return requested_by_id
|
|
if obj.created_by_id:
|
|
return str(obj.created_by_id)
|
|
return None
|
|
|
|
def get_requested_by_name(self, obj):
|
|
return obj.extra.get("requested_by_name")
|
|
|
|
def get_requested_at(self, obj):
|
|
requested_at = obj.extra.get("requested_at")
|
|
if requested_at:
|
|
return requested_at
|
|
if obj.created_at:
|
|
return obj.created_at.isoformat()
|
|
return None
|
|
|
|
def get_status(self, obj):
|
|
issue = obj.issue
|
|
if issue and issue.state and issue.state.group in ["completed", "cancelled"]:
|
|
return "closed"
|
|
return "open"
|
|
|
|
|
|
class ExternalContourBoardItemSerializer(ExternalContourRequestSerializer):
|
|
direction = serializers.SerializerMethodField()
|
|
source_project = serializers.SerializerMethodField()
|
|
target_project = serializers.SerializerMethodField()
|
|
requested_by = serializers.SerializerMethodField()
|
|
capabilities = serializers.SerializerMethodField()
|
|
|
|
class Meta(ExternalContourRequestSerializer.Meta):
|
|
fields = ExternalContourRequestSerializer.Meta.fields + [
|
|
"direction",
|
|
"source_project",
|
|
"target_project",
|
|
"requested_by",
|
|
"capabilities",
|
|
]
|
|
read_only_fields = fields
|
|
|
|
def _resolve_direction(self, obj):
|
|
current_project_id = str(self.context.get("current_project_id") or "")
|
|
source_project_id = str(obj.extra.get("source_project_id") or "")
|
|
target_project_id = str(obj.extra.get("target_project_id") or obj.issue.project_id or "")
|
|
|
|
if current_project_id and current_project_id == source_project_id:
|
|
return "outgoing"
|
|
if current_project_id and current_project_id == target_project_id:
|
|
return "incoming"
|
|
return self.context.get("direction") or "outgoing"
|
|
|
|
def get_has_unread_updates(self, obj):
|
|
unread_request_ids = self.context.get("unread_request_ids")
|
|
if unread_request_ids is not None:
|
|
return str(obj.id) in unread_request_ids
|
|
return super().get_has_unread_updates(obj)
|
|
|
|
def _get_project_payload(self, project_id, fallback_name=None):
|
|
if not project_id:
|
|
return None
|
|
|
|
project = (self.context.get("project_map") or {}).get(str(project_id))
|
|
return {
|
|
"id": str(project_id),
|
|
"identifier": getattr(project, "identifier", None),
|
|
"name": getattr(project, "name", None) or fallback_name,
|
|
"logo_props": getattr(project, "logo_props", None),
|
|
}
|
|
|
|
def get_direction(self, obj):
|
|
return self._resolve_direction(obj)
|
|
|
|
def get_source_project(self, obj):
|
|
source_project_id = obj.extra.get("source_project_id")
|
|
fallback_name = obj.extra.get("source_project_name")
|
|
return self._get_project_payload(source_project_id, fallback_name=fallback_name)
|
|
|
|
def get_target_project(self, obj):
|
|
target_project_id = obj.extra.get("target_project_id") or (str(obj.issue.project_id) if obj.issue_id else None)
|
|
fallback_name = obj.extra.get("target_project_name") or (obj.issue.project.name if obj.issue and obj.issue.project else None)
|
|
return self._get_project_payload(target_project_id, fallback_name=fallback_name)
|
|
|
|
def get_requested_by(self, obj):
|
|
requested_by_id = obj.extra.get("requested_by_id") or (str(obj.created_by_id) if obj.created_by_id else None)
|
|
requested_by_name = obj.extra.get("requested_by_name") or self.get_requested_by_name(obj)
|
|
return {
|
|
"id": requested_by_id,
|
|
"display_name": requested_by_name,
|
|
}
|
|
|
|
def get_capabilities(self, obj):
|
|
request = self.context.get("request")
|
|
user_id = str(getattr(getattr(request, "user", None), "id", "") or "")
|
|
requested_by_id = str(obj.extra.get("requested_by_id") or obj.created_by_id or "")
|
|
direction = self._resolve_direction(obj)
|
|
issue_project_id = str(obj.issue.project_id) if obj.issue_id and obj.issue and obj.issue.project_id else ""
|
|
member_project_ids = self.context.get("member_project_ids") or set()
|
|
is_open_request = self.get_status(obj) == "open"
|
|
|
|
return {
|
|
"can_open_detail": True,
|
|
"can_open_target_issue": issue_project_id in member_project_ids,
|
|
"can_edit_request": direction == "outgoing" and user_id == requested_by_id and is_open_request,
|
|
"can_reply": direction == "outgoing" and bool(obj.issue_id),
|
|
"can_source_decide": direction == "outgoing" and not is_open_request,
|
|
}
|
|
|
|
def get_source_project_name(self, obj):
|
|
source_project = self.get_source_project(obj)
|
|
return source_project["name"] if source_project else None
|
|
|
|
def get_target_project_name(self, obj):
|
|
target_project = self.get_target_project(obj)
|
|
return target_project["name"] if target_project else None
|