oncall-engine/engine/apps/api/views/user_notification_policy.py

from django.conf import settings
from django.http import Http404
from django_filters import rest_framework as filters
from drf_spectacular.utils import extend_schema, extend_schema_view, inline_serializer
from rest_framework import serializers
from rest_framework.decorators import action
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response

from apps.api.permissions import IsOwnerOrHasRBACPermissions, RBACPermission
from apps.api.serializers.user_notification_policy import (
    UserNotificationPolicySerializer,
    UserNotificationPolicyUpdateSerializer,
)
from apps.auth_token.auth import PluginAuthentication
from apps.base.messaging import get_messaging_backend_from_id
from apps.base.models import UserNotificationPolicy
from apps.base.models.user_notification_policy import BUILT_IN_BACKENDS, NotificationChannelAPIOptions
from apps.mobile_app.auth import MobileAppAuthTokenAuthentication
from common.api_helpers.filters import ModelChoicePublicPrimaryKeyFilter, get_user_queryset
from common.api_helpers.mixins import UpdateSerializerMixin
from common.insight_log import EntityEvent, write_resource_insight_log
from common.ordered_model.viewset import OrderedModelViewSet


class UserNotificationPolicyFilter(filters.FilterSet):
    important = filters.BooleanFilter()
    user = ModelChoicePublicPrimaryKeyFilter(
        queryset=get_user_queryset,
    )


@extend_schema_view(
    list=extend_schema(responses=UserNotificationPolicySerializer),
    update=extend_schema(responses=UserNotificationPolicyUpdateSerializer),
    partial_update=extend_schema(responses=UserNotificationPolicyUpdateSerializer),
)
class UserNotificationPolicyView(UpdateSerializerMixin, OrderedModelViewSet):
    """
    Internal API endpoints for user notification policies.
    """

    authentication_classes = (
        MobileAppAuthTokenAuthentication,
        PluginAuthentication,
    )
    permission_classes = (IsAuthenticated, RBACPermission)

    rbac_permissions = {
        "metadata": [RBACPermission.Permissions.USER_SETTINGS_READ],
        "list": [RBACPermission.Permissions.USER_SETTINGS_READ],
        "retrieve": [RBACPermission.Permissions.USER_SETTINGS_READ],
        "delay_options": [RBACPermission.Permissions.USER_SETTINGS_READ],
        "notify_by_options": [RBACPermission.Permissions.USER_SETTINGS_READ],
        "create": [RBACPermission.Permissions.USER_SETTINGS_WRITE],
        "update": [RBACPermission.Permissions.USER_SETTINGS_WRITE],
        "partial_update": [RBACPermission.Permissions.USER_SETTINGS_WRITE],
        "destroy": [RBACPermission.Permissions.USER_SETTINGS_WRITE],
        "move_to_position": [RBACPermission.Permissions.USER_SETTINGS_WRITE],
    }

    IsOwnerOrHasUserSettingsAdminPermission = IsOwnerOrHasRBACPermissions(
        required_permissions=[RBACPermission.Permissions.USER_SETTINGS_ADMIN], ownership_field="user"
    )

    rbac_object_permissions = {
        IsOwnerOrHasUserSettingsAdminPermission: [
            "create",
            "update",
            "partial_update",
            "destroy",
            "move_to_position",
        ],
    }
    queryset = UserNotificationPolicy.objects.none()  # needed for drf-spectacular introspection

    model = UserNotificationPolicy
    serializer_class = UserNotificationPolicySerializer
    update_serializer_class = UserNotificationPolicyUpdateSerializer

    filter_backends = (filters.DjangoFilterBackend,)
    filterset_class = UserNotificationPolicyFilter

    def get_queryset(self):
        # if there are no query params, set default value
        lookup_kwargs = {}
        important = self.request.query_params.get("important", None)
        user_id = self.request.query_params.get("user", None)
        if important is None:
            lookup_kwargs.update({"important": False})
        if user_id is None:
            lookup_kwargs.update({"user": self.request.user})
        queryset = UserNotificationPolicy.objects.filter(
            **lookup_kwargs, user__organization=self.request.auth.organization
        )
        return self.serializer_class.setup_eager_loading(queryset)

    def get_object(self):
        # we need overriden get object, because original one call get_queryset first and raise 404 trying to access
        # other user policies
        pk = self.kwargs["pk"]
        organization = self.request.auth.organization

        try:
            obj = UserNotificationPolicy.objects.get(public_primary_key=pk, user__organization=organization)
        except UserNotificationPolicy.DoesNotExist:
            raise Http404

        self.check_object_permissions(self.request, obj)
        return obj

    def perform_create(self, serializer):
        user = serializer.validated_data.get("user") or self.request.user
        prev_state = user.insight_logs_serialized
        serializer.save()
        new_state = user.insight_logs_serialized
        write_resource_insight_log(
            instance=user,
            author=self.request.user,
            event=EntityEvent.UPDATED,
            prev_state=prev_state,
            new_state=new_state,
        )

    def perform_update(self, serializer):
        user = serializer.validated_data.get("user") or self.request.user
        prev_state = user.insight_logs_serialized
        serializer.save()
        new_state = user.insight_logs_serialized
        write_resource_insight_log(
            instance=user,
            author=self.request.user,
            event=EntityEvent.UPDATED,
            prev_state=prev_state,
            new_state=new_state,
        )

    def perform_destroy(self, instance):
        user = instance.user
        prev_state = user.insight_logs_serialized
        instance.delete()
        new_state = user.insight_logs_serialized
        write_resource_insight_log(
            instance=user,
            author=self.request.user,
            event=EntityEvent.UPDATED,
            prev_state=prev_state,
            new_state=new_state,
        )

    @extend_schema(
        responses=inline_serializer(
            name="UserNotificationPolicyDelayOptions",
            fields={
                "value": serializers.CharField(),
                "sec_value": serializers.IntegerField(),
                "display_name": serializers.CharField(),
            },
            many=True,
        )
    )
    @action(detail=False, methods=["get"])
    def delay_options(self, request):
        choices = []
        for item in UserNotificationPolicy.DURATION_CHOICES:
            choices.append({"value": str(item[0]), "sec_value": item[0], "display_name": item[1]})
        return Response(choices)

    @extend_schema(
        responses=inline_serializer(
            name="UserNotificationPolicyNotifyByOptions",
            fields={
                "value": serializers.IntegerField(),
                "display_name": serializers.CharField(),
                "slack_integration_required": serializers.BooleanField(),
                "telegram_integration_required": serializers.BooleanField(),
            },
            many=True,
        )
    )
    @action(detail=False, methods=["get"])
    def notify_by_options(self, request):
        """
        Returns list of options for user notification policies dropping options that requires disabled features.
        """
        choices = []
        for notification_channel in NotificationChannelAPIOptions.AVAILABLE_FOR_USE:
            slack_integration_required = (
                notification_channel in NotificationChannelAPIOptions.SLACK_INTEGRATION_REQUIRED_NOTIFICATION_CHANNELS
            )
            telegram_integration_required = (
                notification_channel
                in NotificationChannelAPIOptions.TELEGRAM_INTEGRATION_REQUIRED_NOTIFICATION_CHANNELS
            )
            if slack_integration_required and not settings.FEATURE_SLACK_INTEGRATION_ENABLED:
                continue
            if telegram_integration_required and not settings.FEATURE_TELEGRAM_INTEGRATION_ENABLED:
                continue

            # extra backends may be enabled per organization
            built_in_backend_names = {b[0] for b in BUILT_IN_BACKENDS}
            if notification_channel.name not in built_in_backend_names:
                extra_messaging_backend = get_messaging_backend_from_id(notification_channel.name)
                if extra_messaging_backend is None or not extra_messaging_backend.is_enabled_for_organization(
                    request.auth.organization
                ):
                    continue

            choices.append(
                {
                    "value": notification_channel,
                    "display_name": NotificationChannelAPIOptions.LABELS[notification_channel],
                    "slack_integration_required": slack_integration_required,
                    "telegram_integration_required": telegram_integration_required,
                }
            )
        return Response(choices)