You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
115 lines
4.7 KiB
115 lines
4.7 KiB
from rest_framework import generics, status
|
|
from rest_framework.response import Response
|
|
from rest_framework.authentication import TokenAuthentication
|
|
from drf_yasg.utils import swagger_auto_schema
|
|
from drf_yasg import openapi
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.viewsets import ModelViewSet
|
|
|
|
from apps.account.serializers import (
|
|
NotificationSerializer,
|
|
NotificationSendSerializer,
|
|
AdminNotificationSerializer,
|
|
NotificationTemplateSerializer
|
|
)
|
|
from apps.account.models import Notification, User, NotificationTemplate
|
|
from apps.account.tasks import send_notification_task
|
|
from utils.pagination import StandardResultsSetPagination
|
|
from apps.account.permissions import IsSuperAdmin
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
from drf_spectacular.utils import extend_schema, OpenApiResponse, inline_serializer
|
|
from rest_framework import serializers
|
|
|
|
class NotificationListView(generics.ListAPIView):
|
|
serializer_class = NotificationSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
authentication_classes = [TokenAuthentication]
|
|
pagination_class = StandardResultsSetPagination
|
|
|
|
@extend_schema(
|
|
summary="Retrieve user notifications",
|
|
description="Retrieve a paginated list of notifications for the authenticated user.",
|
|
tags=['Notifications'],
|
|
responses={200: NotificationSerializer(many=True)},
|
|
)
|
|
def get(self, request, *args, **kwargs):
|
|
return super().get(request, *args, **kwargs)
|
|
|
|
def get_queryset(self):
|
|
if getattr(self, 'swagger_fake_view', False) or not self.request.user.is_authenticated:
|
|
return Notification.objects.none()
|
|
return Notification.objects.filter(user=self.request.user).order_by('-created_at')
|
|
|
|
|
|
class NotificationReadAllView(generics.GenericAPIView):
|
|
permission_classes = [IsAuthenticated]
|
|
authentication_classes = [TokenAuthentication]
|
|
serializer_class = serializers.Serializer
|
|
|
|
@extend_schema(
|
|
summary="Mark all notifications as read",
|
|
description="Mark all notifications as read for the authenticated user.",
|
|
tags=['Notifications'],
|
|
request=None,
|
|
responses={200: OpenApiResponse(description="All notifications marked as read")}
|
|
)
|
|
def post(self, request, *args, **kwargs):
|
|
Notification.objects.filter(user=request.user, is_read=False).update(is_read=True)
|
|
return Response({'status': 'all notifications marked as read'}, status=status.HTTP_200_OK)
|
|
|
|
|
|
class SendNotificationSerializer(serializers.Serializer):
|
|
user_id = serializers.IntegerField(help_text='Target User ID')
|
|
title = serializers.CharField(max_length=255, help_text='Notification title')
|
|
body = serializers.CharField(help_text='Notification body')
|
|
data = serializers.DictField(required=False, default=dict, help_text='Extra payload data')
|
|
|
|
|
|
class SendNotificationView(generics.GenericAPIView):
|
|
permission_classes = [IsAuthenticated]
|
|
authentication_classes = [TokenAuthentication]
|
|
serializer_class = SendNotificationSerializer
|
|
|
|
@extend_schema(
|
|
summary="Send notification to user",
|
|
description="Dispatch an in-app notification to a specific user.",
|
|
tags=['Notifications'],
|
|
request=SendNotificationSerializer,
|
|
responses={
|
|
200: OpenApiResponse(description='Notification dispatched.'),
|
|
404: OpenApiResponse(description='User not found.'),
|
|
}
|
|
)
|
|
def post(self, request, *args, **kwargs):
|
|
serializer = self.get_serializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
user_id = serializer.validated_data.get('user_id')
|
|
title = serializer.validated_data.get('title')
|
|
body = serializer.validated_data.get('body')
|
|
data = serializer.validated_data.get('data', {})
|
|
|
|
if not User.objects.filter(id=user_id).exists():
|
|
return Response({'error': 'User not found.'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
send_notification_task.delay(user_id=user_id, title=title, body=body, data=data)
|
|
return Response({'status': 'notification queued'}, status=status.HTTP_200_OK)
|
|
|
|
|
|
class AdminNotificationViewSet(ModelViewSet):
|
|
queryset = Notification.objects.all().select_related('user')
|
|
serializer_class = AdminNotificationSerializer
|
|
permission_classes = [IsAuthenticated, IsSuperAdmin]
|
|
authentication_classes = [TokenAuthentication]
|
|
pagination_class = StandardResultsSetPagination
|
|
|
|
|
|
class AdminNotificationTemplateViewSet(ModelViewSet):
|
|
queryset = NotificationTemplate.objects.all()
|
|
serializer_class = NotificationTemplateSerializer
|
|
permission_classes = [IsAuthenticated, IsSuperAdmin]
|
|
authentication_classes = [TokenAuthentication]
|
|
pagination_class = StandardResultsSetPagination
|