133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
"""
|
|
Notification Views
|
|
"""
|
|
|
|
from rest_framework import viewsets, status
|
|
from rest_framework.decorators import action
|
|
from rest_framework.response import Response
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from django.utils import timezone
|
|
|
|
from igny8_core.api.pagination import CustomPageNumberPagination
|
|
from igny8_core.api.base import AccountModelViewSet
|
|
|
|
from .models import Notification
|
|
from .serializers import NotificationSerializer, NotificationListSerializer, MarkReadSerializer
|
|
|
|
|
|
class NotificationViewSet(AccountModelViewSet):
|
|
"""
|
|
ViewSet for managing notifications
|
|
|
|
Endpoints:
|
|
- GET /api/v1/notifications/ - List notifications
|
|
- GET /api/v1/notifications/{id}/ - Get notification detail
|
|
- DELETE /api/v1/notifications/{id}/ - Delete notification
|
|
- POST /api/v1/notifications/{id}/read/ - Mark single notification as read
|
|
- POST /api/v1/notifications/read-all/ - Mark all notifications as read
|
|
- GET /api/v1/notifications/unread-count/ - Get unread notification count
|
|
"""
|
|
|
|
serializer_class = NotificationSerializer
|
|
pagination_class = CustomPageNumberPagination
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get_queryset(self):
|
|
"""Filter notifications for current account and user"""
|
|
from django.db.models import Q
|
|
|
|
user = self.request.user
|
|
account = getattr(user, 'account', None)
|
|
|
|
if not account:
|
|
return Notification.objects.none()
|
|
|
|
# Get notifications for this account that are either:
|
|
# - For all users (user=None)
|
|
# - For this specific user
|
|
queryset = Notification.objects.filter(
|
|
Q(account=account, user__isnull=True) |
|
|
Q(account=account, user=user)
|
|
).select_related('site').order_by('-created_at')
|
|
|
|
# Optional filters
|
|
is_read = self.request.query_params.get('is_read')
|
|
if is_read is not None:
|
|
queryset = queryset.filter(is_read=is_read.lower() == 'true')
|
|
|
|
notification_type = self.request.query_params.get('type')
|
|
if notification_type:
|
|
queryset = queryset.filter(notification_type=notification_type)
|
|
|
|
severity = self.request.query_params.get('severity')
|
|
if severity:
|
|
queryset = queryset.filter(severity=severity)
|
|
|
|
return queryset
|
|
|
|
def get_serializer_class(self):
|
|
"""Use list serializer for list action"""
|
|
if self.action == 'list':
|
|
return NotificationListSerializer
|
|
return NotificationSerializer
|
|
|
|
def list(self, request, *args, **kwargs):
|
|
"""List notifications with unread count"""
|
|
queryset = self.filter_queryset(self.get_queryset())
|
|
|
|
# Get unread count
|
|
unread_count = queryset.filter(is_read=False).count()
|
|
|
|
page = self.paginate_queryset(queryset)
|
|
if page is not None:
|
|
serializer = self.get_serializer(page, many=True)
|
|
response = self.get_paginated_response(serializer.data)
|
|
response.data['unread_count'] = unread_count
|
|
return response
|
|
|
|
serializer = self.get_serializer(queryset, many=True)
|
|
return Response({
|
|
'results': serializer.data,
|
|
'unread_count': unread_count
|
|
})
|
|
|
|
@action(detail=True, methods=['post'])
|
|
def read(self, request, pk=None):
|
|
"""Mark a single notification as read"""
|
|
notification = self.get_object()
|
|
notification.mark_as_read()
|
|
serializer = self.get_serializer(notification)
|
|
return Response(serializer.data)
|
|
|
|
@action(detail=False, methods=['post'], url_path='read-all')
|
|
def read_all(self, request):
|
|
"""Mark all notifications as read"""
|
|
serializer = MarkReadSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
|
|
notification_ids = serializer.validated_data.get('notification_ids', [])
|
|
|
|
queryset = self.get_queryset().filter(is_read=False)
|
|
|
|
if notification_ids:
|
|
queryset = queryset.filter(id__in=notification_ids)
|
|
|
|
count = queryset.update(is_read=True, read_at=timezone.now())
|
|
|
|
return Response({
|
|
'status': 'success',
|
|
'marked_read': count
|
|
})
|
|
|
|
@action(detail=False, methods=['get'], url_path='unread-count')
|
|
def unread_count(self, request):
|
|
"""Get count of unread notifications"""
|
|
count = self.get_queryset().filter(is_read=False).count()
|
|
return Response({'unread_count': count})
|
|
|
|
def destroy(self, request, *args, **kwargs):
|
|
"""Delete a notification"""
|
|
instance = self.get_object()
|
|
self.perform_destroy(instance)
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|