58 lines
2.3 KiB
Python
58 lines
2.3 KiB
Python
from rest_framework import viewsets, status
|
|
from rest_framework.decorators import action
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.exceptions import PermissionDenied
|
|
from rest_framework.response import Response
|
|
|
|
from .models import Notificacion, TipoNotificacion
|
|
from .serializers import NotificacionSerializer, TipoNotificacionSerializer
|
|
from core.permissions import require_permission
|
|
|
|
|
|
class TipoNotificacionViewSet(viewsets.ModelViewSet):
|
|
queryset = TipoNotificacion.objects.all()
|
|
serializer_class = TipoNotificacionSerializer
|
|
http_method_names = ['get']
|
|
permission_classes = [IsAuthenticated]
|
|
my_tags = ['Notificaciones']
|
|
|
|
def get_queryset(self):
|
|
return self.queryset.order_by('tipo')
|
|
|
|
|
|
class NotificacionViewSet(viewsets.ModelViewSet):
|
|
queryset = Notificacion.objects.all()
|
|
serializer_class = NotificacionSerializer
|
|
http_method_names = ['get', 'post', 'put', 'patch', 'delete']
|
|
filterset_fields = ['visto']
|
|
my_tags = ['Notificaciones']
|
|
|
|
def get_permissions(self):
|
|
if self.action in ('list', 'retrieve'):
|
|
return [IsAuthenticated(), require_permission('notificaciones.view')()]
|
|
return [IsAuthenticated()]
|
|
|
|
def get_queryset(self):
|
|
if getattr(self, 'swagger_fake_view', False):
|
|
return Notificacion.objects.none()
|
|
user = self.request.user
|
|
if not user.is_authenticated:
|
|
return Notificacion.objects.none()
|
|
return Notificacion.objects.filter(dirigido=user)
|
|
|
|
def perform_create(self, serializer):
|
|
if not self.request.user.is_authenticated:
|
|
raise PermissionDenied("Usuario no autenticado")
|
|
if self.request.user.is_superuser:
|
|
serializer.save()
|
|
return
|
|
raise PermissionDenied("No tienes permiso para crear notificaciones para otros usuarios")
|
|
|
|
@action(detail=False, methods=['get'], url_path=r'by-task/(?P<task_id>[^/.]+)')
|
|
def by_task(self, request, task_id=None):
|
|
"""Recupera la notificación de una tarea de auditoría por su task_id (Celery)."""
|
|
notif = self.get_queryset().filter(datos__task_id=task_id).first()
|
|
if not notif:
|
|
return Response({'detail': 'No encontrada.'}, status=status.HTTP_404_NOT_FOUND)
|
|
return Response(self.get_serializer(notif).data)
|