# Copyright (c) 2023-present Plane Software, Inc. and contributors # SPDX-License-Identifier: AGPL-3.0-only # See the LICENSE file for details. # Python imports import json import uuid # Django imports from django.utils import timezone from django.core.serializers.json import DjangoJSONEncoder from django.conf import settings # Third Party imports from rest_framework.response import Response from rest_framework import status from rest_framework.parsers import MultiPartParser, FormParser # Module imports from .. import BaseAPIView from plane.app.serializers import IssueAttachmentSerializer from plane.db.models import FileAsset, Project, Workspace from plane.bgtasks.issue_activities_task import issue_activity from plane.app.permissions import allow_permission, ROLE from plane.settings.storage import S3Storage from plane.utils.host import base_host from plane.utils.upload_limits import get_project_storage_quota_response, resolve_workspace_upload_size_limit from plane.utils.attachment_preview import attachment_object_exists, get_attachment_preview_response from plane.utils.file_dedup import finalize_uploaded_file_asset, release_file_asset_blob, UploadedObjectMissing class IssueAttachmentEndpoint(BaseAPIView): serializer_class = IssueAttachmentSerializer model = FileAsset parser_classes = (MultiPartParser, FormParser) @allow_permission([ROLE.ADMIN, ROLE.MEMBER]) def post(self, request, slug, project_id, issue_id): serializer = IssueAttachmentSerializer(data=request.data) workspace = Workspace.objects.get(slug=slug) if serializer.is_valid(): serializer.save( project_id=project_id, issue_id=issue_id, workspace_id=workspace.id, entity_type=FileAsset.EntityTypeContext.ISSUE_ATTACHMENT, ) issue_activity.delay( type="attachment.activity.created", requested_data=None, actor_id=str(self.request.user.id), issue_id=str(self.kwargs.get("issue_id", None)), project_id=str(self.kwargs.get("project_id", None)), current_instance=json.dumps(serializer.data, cls=DjangoJSONEncoder), epoch=int(timezone.now().timestamp()), notification=True, origin=base_host(request=request, is_app=True), ) return Response(serializer.data, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @allow_permission([ROLE.ADMIN], creator=True, model=FileAsset) def delete(self, request, slug, project_id, issue_id, pk): issue_attachment = FileAsset.objects.filter( pk=pk, workspace__slug=slug, project_id=project_id, issue_id=issue_id ).first() if not issue_attachment: return Response( {"error": "Issue attachment not found."}, status=status.HTTP_404_NOT_FOUND, ) release_result = release_file_asset_blob(issue_attachment, request=request) if not release_result.had_blob: issue_attachment.asset.delete(save=False) issue_attachment.delete() issue_activity.delay( type="attachment.activity.deleted", requested_data=None, actor_id=str(self.request.user.id), issue_id=str(self.kwargs.get("issue_id", None)), project_id=str(self.kwargs.get("project_id", None)), current_instance=None, epoch=int(timezone.now().timestamp()), notification=True, origin=base_host(request=request, is_app=True), ) return Response(status=status.HTTP_204_NO_CONTENT) @allow_permission([ROLE.ADMIN, ROLE.MEMBER, ROLE.GUEST]) def get(self, request, slug, project_id, issue_id): issue_attachments = FileAsset.objects.filter( issue_id=issue_id, workspace__slug=slug, project_id=project_id, entity_type=FileAsset.EntityTypeContext.ISSUE_ATTACHMENT, is_uploaded=True, ) serializer = IssueAttachmentSerializer(issue_attachments, many=True) return Response(serializer.data, status=status.HTTP_200_OK) class IssueAttachmentV2Endpoint(BaseAPIView): serializer_class = IssueAttachmentSerializer model = FileAsset @allow_permission([ROLE.ADMIN, ROLE.MEMBER]) def post(self, request, slug, project_id, issue_id): name = request.data.get("name") type = request.data.get("type", False) size = int(request.data.get("size", settings.FILE_SIZE_LIMIT)) beam_viewer = request.data.get("beamViewer") is_beam_viewer_reference = ( isinstance(beam_viewer, dict) and beam_viewer.get("src") and beam_viewer.get("downloadUrl") ) if not type or type not in settings.ATTACHMENT_MIME_TYPES: return Response( {"error": "Invalid file type.", "status": False}, status=status.HTTP_400_BAD_REQUEST, ) # Get the workspace workspace = Workspace.objects.get(slug=slug) project = Project.objects.get(id=project_id, workspace=workspace) if is_beam_viewer_reference: asset = FileAsset.objects.create( attributes={ "name": name, "type": type, "size": size, "beamViewer": beam_viewer, }, asset=f"{workspace.id}/beam-viewer/{uuid.uuid4().hex}-{name}", size=0, workspace_id=workspace.id, created_by=request.user, issue_id=issue_id, project_id=project_id, entity_type=FileAsset.EntityTypeContext.ISSUE_ATTACHMENT, is_uploaded=True, external_source="beam-viewer", ) serializer = IssueAttachmentSerializer(asset) issue_activity.delay( type="attachment.activity.created", requested_data=None, actor_id=str(self.request.user.id), issue_id=str(self.kwargs.get("issue_id", None)), project_id=str(self.kwargs.get("project_id", None)), current_instance=json.dumps(serializer.data, cls=DjangoJSONEncoder), epoch=int(timezone.now().timestamp()), notification=True, origin=base_host(request=request, is_app=True), ) return Response( { "asset_id": str(asset.id), "attachment": serializer.data, "asset_url": asset.asset_url, }, status=status.HTTP_200_OK, ) # asset key asset_key = f"{workspace.id}/{uuid.uuid4().hex}-{name}" # Get the size limit size_limit = resolve_workspace_upload_size_limit(workspace, size) quota_response = get_project_storage_quota_response(project, requested_size=size) if quota_response is not None: return quota_response # Create a File Asset asset = FileAsset.objects.create( attributes={"name": name, "type": type, "size": size_limit}, asset=asset_key, size=size_limit, workspace_id=workspace.id, created_by=request.user, issue_id=issue_id, project_id=project_id, entity_type=FileAsset.EntityTypeContext.ISSUE_ATTACHMENT, ) # Get the presigned URL storage = S3Storage(request=request) # Generate a presigned URL to share an S3 object presigned_url = storage.generate_presigned_post(object_name=asset_key, file_type=type, file_size=size_limit) # Return the presigned URL return Response( { "upload_data": presigned_url, "asset_id": str(asset.id), "attachment": IssueAttachmentSerializer(asset).data, "asset_url": asset.asset_url, }, status=status.HTTP_200_OK, ) @allow_permission([ROLE.ADMIN], creator=True, model=FileAsset) def delete(self, request, slug, project_id, issue_id, pk): issue_attachment = FileAsset.objects.get(pk=pk, workspace__slug=slug, project_id=project_id) if not issue_attachment.is_uploaded: release_file_asset_blob(issue_attachment, request=request, delete_untracked_object=True) issue_attachment.is_deleted = True issue_attachment.deleted_at = timezone.now() issue_attachment.save(update_fields=["is_deleted", "deleted_at", "updated_at"]) issue_activity.delay( type="attachment.activity.deleted", requested_data=None, actor_id=str(self.request.user.id), issue_id=str(issue_id), project_id=str(project_id), current_instance=None, epoch=int(timezone.now().timestamp()), notification=True, origin=base_host(request=request, is_app=True), ) return Response(status=status.HTTP_204_NO_CONTENT) @allow_permission([ROLE.ADMIN, ROLE.MEMBER, ROLE.GUEST]) def get(self, request, slug, project_id, issue_id, pk=None): if pk: # Get the asset asset = FileAsset.objects.get(id=pk, workspace__slug=slug, project_id=project_id) # Check if the asset is uploaded if not asset.is_uploaded: return Response( {"error": "The asset is not uploaded.", "status": False}, status=status.HTTP_400_BAD_REQUEST, ) disposition = "inline" if request.GET.get("preview") == "true" else "attachment" return get_attachment_preview_response(request, asset, disposition=disposition) # Get all the attachments issue_attachments = FileAsset.objects.filter( issue_id=issue_id, entity_type=FileAsset.EntityTypeContext.ISSUE_ATTACHMENT, workspace__slug=slug, project_id=project_id, is_uploaded=True, ) # Serialize the attachments serializer = IssueAttachmentSerializer(issue_attachments, many=True) return Response(serializer.data, status=status.HTTP_200_OK) @allow_permission([ROLE.ADMIN, ROLE.MEMBER]) def patch(self, request, slug, project_id, issue_id, pk): issue_attachment = FileAsset.objects.get(pk=pk, workspace__slug=slug, project_id=project_id) beam_viewer = request.data.get("beamViewer") if isinstance(beam_viewer, dict): attributes = issue_attachment.attributes or {} existing_beam_viewer = attributes.get("beamViewer") if not isinstance(existing_beam_viewer, dict): return Response( {"error": "The attachment is not a Beam Viewer reference.", "status": False}, status=status.HTTP_400_BAD_REQUEST, ) attributes["beamViewer"] = { **existing_beam_viewer, **beam_viewer, } issue_attachment.attributes = attributes issue_attachment.is_uploaded = True issue_attachment.save(update_fields=["attributes", "is_uploaded", "updated_at"]) return Response(IssueAttachmentSerializer(issue_attachment).data, status=status.HTTP_200_OK) serializer = IssueAttachmentSerializer(issue_attachment) if not attachment_object_exists(issue_attachment): return Response( {"error": "The uploaded attachment object was not found.", "status": False}, status=status.HTTP_400_BAD_REQUEST, ) # Send this activity only if the attachment is not uploaded before if not issue_attachment.is_uploaded: issue_activity.delay( type="attachment.activity.created", requested_data=None, actor_id=str(self.request.user.id), issue_id=str(self.kwargs.get("issue_id", None)), project_id=str(self.kwargs.get("project_id", None)), current_instance=json.dumps(serializer.data, cls=DjangoJSONEncoder), epoch=int(timezone.now().timestamp()), notification=True, origin=base_host(request=request, is_app=True), ) issue_attachment.created_by = request.user issue_attachment.save(update_fields=["created_by", "updated_at"]) try: finalize_uploaded_file_asset(issue_attachment, request=request) except UploadedObjectMissing: return Response( {"error": "The uploaded attachment object was not found.", "status": False}, status=status.HTTP_400_BAD_REQUEST, ) return Response(status=status.HTTP_204_NO_CONTENT)