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.
103 lines
4.1 KiB
103 lines
4.1 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__)
|
|
|
|
|
|
class NotificationListView(generics.ListAPIView):
|
|
serializer_class = NotificationSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
authentication_classes = [TokenAuthentication]
|
|
pagination_class = StandardResultsSetPagination
|
|
|
|
@swagger_auto_schema(
|
|
operation_description="Retrieve a list of notifications for the authenticated user.",
|
|
tags=['Notifications'],
|
|
)
|
|
def get(self, request, *args, **kwargs):
|
|
return super().get(request, *args, **kwargs)
|
|
|
|
def get_queryset(self):
|
|
return Notification.objects.filter(user=self.request.user).order_by('-created_at')
|
|
|
|
|
|
class NotificationReadAllView(generics.GenericAPIView):
|
|
permission_classes = [IsAuthenticated]
|
|
authentication_classes = [TokenAuthentication]
|
|
|
|
@swagger_auto_schema(
|
|
operation_description="Mark all notifications as read for the authenticated user.",
|
|
tags=['Notifications'],
|
|
responses={200: "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 SendNotificationView(generics.GenericAPIView):
|
|
permission_classes = [IsAuthenticated]
|
|
authentication_classes = [TokenAuthentication]
|
|
|
|
@swagger_auto_schema(
|
|
operation_description="Dispatch an in-app notification to a specific user.",
|
|
tags=['Notifications'],
|
|
request_body=openapi.Schema(
|
|
type=openapi.TYPE_OBJECT,
|
|
required=['user_id', 'title', 'body'],
|
|
properties={
|
|
'user_id': openapi.Schema(type=openapi.TYPE_INTEGER, description='Target User ID'),
|
|
'title': openapi.Schema(type=openapi.TYPE_STRING, description='Notification title'),
|
|
'body': openapi.Schema(type=openapi.TYPE_STRING, description='Notification body'),
|
|
'data': openapi.Schema(type=openapi.TYPE_OBJECT, description='Extra payload data'),
|
|
},
|
|
),
|
|
responses={
|
|
200: openapi.Response('Notification dispatched.'),
|
|
404: openapi.Response('User not found.'),
|
|
}
|
|
)
|
|
def post(self, request, *args, **kwargs):
|
|
user_id = request.data.get('user_id')
|
|
title = request.data.get('title')
|
|
body = request.data.get('body')
|
|
data = request.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
|