NODEDC_TASKMANAGER/plane-src/apps/api/plane/api/serializers/external_contours.py

295 lines
11 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, 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 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)
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):
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",
"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_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):
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"