insights/products/approvals/backend/api.py

import logging
from typing import Any, cast

from django.db.models import QuerySet

import django_filters
from django_filters.rest_framework import DjangoFilterBackend
from drf_spectacular.utils import extend_schema
from rest_framework import exceptions, status, viewsets
from rest_framework.decorators import action
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.serializers import BaseSerializer

from insights.api.routing import TeamAndOrgViewSetMixin
from insights.constants import AvailableFeature
from insights.models import User
from insights.permissions import (
    OrganizationAdminWritePermissions,
    OrganizationMemberPermissions,
    PremiumFeaturePermission,
)

from products.approvals.backend.exceptions import AlreadyVotedError, InvalidStateError, ReasonRequiredError
from products.approvals.backend.models import ApprovalPolicy, ChangeRequest
from products.approvals.backend.permissions import CanApprove, CanCancel
from products.approvals.backend.serializers import (
    ApprovalPolicySerializer,
    ChangeRequestApproveSerializer,
    ChangeRequestDecisionResponseSerializer,
    ChangeRequestRejectSerializer,
    ChangeRequestSerializer,
)
from products.approvals.backend.services import ChangeRequestService

logger = logging.getLogger(__name__)


class CharInFilter(django_filters.BaseInFilter, django_filters.CharFilter):
    pass


class ChangeRequestFilterSet(django_filters.FilterSet):
    state = CharInFilter(field_name="state", lookup_expr="in")
    action_key = django_filters.CharFilter(field_name="action_key")
    requester = django_filters.NumberFilter(field_name="created_by_id")
    resource_type = django_filters.CharFilter(field_name="resource_type")
    resource_id = django_filters.CharFilter(field_name="resource_id")

    class Meta:
        model = ChangeRequest
        fields: list[str] = []


@extend_schema(tags=["platform_features"])
class ChangeRequestViewSet(TeamAndOrgViewSetMixin, viewsets.ReadOnlyModelViewSet):
    scope_object = "approvals"
    scope_object_write_actions = ["approve", "reject", "cancel"]
    queryset = ChangeRequest.objects.all().order_by("-created_at")
    permission_classes = [OrganizationMemberPermissions, PremiumFeaturePermission]
    premium_feature_on_cloud = AvailableFeature.APPROVALS
    serializer_class = ChangeRequestSerializer
    filter_backends = [DjangoFilterBackend]
    filterset_class = ChangeRequestFilterSet

    def safely_get_queryset(self, queryset: QuerySet) -> QuerySet:
        return queryset.select_related("created_by", "applied_by").prefetch_related("approvals")

    @extend_schema(
        request=ChangeRequestApproveSerializer,
        responses={200: ChangeRequestDecisionResponseSerializer},
    )
    @action(methods=["POST"], detail=True, permission_classes=[PremiumFeaturePermission, CanApprove])
    def approve(self, request: Request, pk=None, **kwargs) -> Response:
        """
        Approve a change request.
        If quorum is reached, automatically applies the change immediately.
        """
        change_request: ChangeRequest = self.get_object()
        service = ChangeRequestService(change_request, cast(User, request.user), request=request)

        try:
            result = service.approve(reason=request.data.get("reason", ""))
        except InvalidStateError:
            return Response(
                {"error": "This change request can no longer be approved."}, status=status.HTTP_400_BAD_REQUEST
            )
        except AlreadyVotedError:
            return Response(
                {"error": "You have already voted on this change request."}, status=status.HTTP_400_BAD_REQUEST
            )
        except Exception as e:
            logger.error("Unexpected error in approve: %s", str(e), exc_info=True)
            return Response(
                {"error": "An error occurred while processing approval."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR
            )

        response_data: dict[str, Any] = {
            "status": result.status,
            "message": result.message,
            "change_request": ChangeRequestSerializer(result.change_request, context={"request": request}).data,
        }
        if result.result_data:
            response_data["result"] = result.result_data

        return Response(response_data, status=status.HTTP_200_OK)

    @extend_schema(
        request=ChangeRequestRejectSerializer,
        responses={200: ChangeRequestDecisionResponseSerializer},
    )
    @action(methods=["POST"], detail=True, permission_classes=[PremiumFeaturePermission, CanApprove])
    def reject(self, request: Request, pk=None, **kwargs) -> Response:
        """Reject a change request."""
        change_request: ChangeRequest = self.get_object()
        service = ChangeRequestService(change_request, cast(User, request.user), request=request)

        try:
            result = service.reject(reason=request.data.get("reason", ""))
        except InvalidStateError:
            return Response(
                {"error": "This change request can no longer be rejected."}, status=status.HTTP_400_BAD_REQUEST
            )
        except ReasonRequiredError:
            return Response({"error": "A reason is required for rejection."}, status=status.HTTP_400_BAD_REQUEST)
        except AlreadyVotedError:
            return Response(
                {"error": "You have already voted on this change request."}, status=status.HTTP_400_BAD_REQUEST
            )
        except Exception as e:
            logger.error("Unexpected error in reject: %s", str(e), exc_info=True)
            return Response(
                {"error": "An error occurred while processing rejection."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR
            )

        return Response(
            {
                "status": result.status,
                "message": result.message,
                "change_request": ChangeRequestSerializer(result.change_request, context={"request": request}).data,
            },
            status=status.HTTP_200_OK,
        )

    @action(methods=["POST"], detail=True, permission_classes=[PremiumFeaturePermission, CanCancel])
    def cancel(self, request: Request, pk=None, **kwargs) -> Response:
        """
        Cancel a change request.
        Only the requester can cancel their own pending change request.
        """
        change_request: ChangeRequest = self.get_object()
        service = ChangeRequestService(change_request, cast(User, request.user), request=request)

        try:
            result = service.cancel(reason=request.data.get("reason", "Canceled by requester"))
        except InvalidStateError:
            return Response(
                {"error": "This change request can no longer be canceled."}, status=status.HTTP_400_BAD_REQUEST
            )
        except Exception as e:
            logger.error("Unexpected error in cancel: %s", str(e), exc_info=True)
            return Response(
                {"error": "An error occurred while canceling the request."},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR,
            )

        return Response(
            {
                "status": result.status,
                "message": result.message,
                "change_request": ChangeRequestSerializer(result.change_request, context={"request": request}).data,
            },
            status=status.HTTP_200_OK,
        )


@extend_schema(tags=["platform_features"])
class ApprovalPolicyViewSet(TeamAndOrgViewSetMixin, viewsets.ModelViewSet):
    scope_object = "approvals"
    queryset = ApprovalPolicy.objects.all().order_by("-created_at")
    serializer_class = ApprovalPolicySerializer
    permission_classes = [OrganizationMemberPermissions, OrganizationAdminWritePermissions, PremiumFeaturePermission]
    premium_feature_on_cloud = AvailableFeature.APPROVALS

    def safely_get_queryset(self, queryset: QuerySet) -> QuerySet:
        filters = self.request.query_params

        if "action_key" in filters:
            queryset = queryset.filter(action_key=filters["action_key"])

        if "enabled" in filters:
            enabled = filters["enabled"].lower() == "true"
            queryset = queryset.filter(enabled=enabled)

        return queryset.select_related("created_by")

    def perform_create(self, serializer: BaseSerializer) -> None:
        serializer.save(
            created_by=self.request.user,
            organization=self.organization,
            team=self.team,
        )

    def create(self, request: Request, *args, **kwargs) -> Response:
        action_key = request.data.get("action_key")

        if (
            action_key
            and ApprovalPolicy.objects.filter(
                action_key=action_key,
                organization=self.organization,
                team=self.team,
            ).exists()
        ):
            raise exceptions.ValidationError(
                "A policy for this action already exists. You can edit the existing policy instead."
            )

        return super().create(request, *args, **kwargs)