Merge pull request #54 from SELab-2/notifications

Notifications

backend/authentication/serializers.py

@@ -1,27 +1,28 @@
 from typing import Tuple
-from django.contrib.auth.models import update_last_login
+
+from authentication.cas.client import client
+from authentication.models import User
+from authentication.signals import user_created, user_login
 from django.contrib.auth import login
+from django.contrib.auth.models import update_last_login
 from rest_framework.serializers import (
     CharField,
     EmailField,
+    HyperlinkedRelatedField,
     ModelSerializer,
-    ValidationError,
     Serializer,
-    HyperlinkedIdentityField,
-    HyperlinkedRelatedField,
+    ValidationError,
 )
-from rest_framework_simplejwt.tokens import RefreshToken, AccessToken
 from rest_framework_simplejwt.settings import api_settings
-from authentication.signals import user_created, user_login
-from authentication.models import User, Faculty
-from authentication.cas.client import client
+from rest_framework_simplejwt.tokens import AccessToken, RefreshToken
 
 
 class CASTokenObtainSerializer(Serializer):
     """Serializer for CAS ticket validation
     This serializer takes the CAS ticket and tries to validate it.
     Upon successful validation, create a new user if it doesn't exist.
     """
+
     ticket = CharField(required=True, min_length=49, max_length=49)
 
     def validate(self, data):
@@ -40,23 +41,15 @@ def validate(self, data):
         if "request" in self.context:
             login(self.context["request"], user)
 
-        user_login.send(
-            sender=self, user=user
-        )
+        user_login.send(sender=self, user=user)
 
         if created:
-            user_created.send(
-                sender=self, attributes=attributes, user=user
-            )
+            user_created.send(sender=self, attributes=attributes, user=user)
 
         # Return access tokens for the now logged-in user.
         return {
-            "access": str(
-                AccessToken.for_user(user)
-            ),
-            "refresh": str(
-                RefreshToken.for_user(user)
-            ),
+            "access": str(AccessToken.for_user(user)),
+            "refresh": str(RefreshToken.for_user(user)),
         }
 
     def _validate_ticket(self, ticket: str) -> dict:
@@ -102,7 +95,7 @@ class UserSerializer(ModelSerializer):
         many=True, read_only=True, view_name="faculty-detail"
     )
 
-    notifications = HyperlinkedIdentityField(
+    notifications = HyperlinkedRelatedField(
         view_name="notification-detail",
         read_only=True,
     )

backend/notifications/apps.py

@@ -4,9 +4,3 @@
 class NotificationsConfig(AppConfig):
     default_auto_field = "django.db.models.BigAutoField"
     name = "notifications"
-
-
-# TODO: Allow is_sent to be adjusted
-# TODO: Signals to send notifications
-# TODO: Send emails
-# TODO: Think about the required api endpoints

backend/notifications/models.py

@@ -4,23 +4,21 @@
 
 class NotificationTemplate(models.Model):
     id = models.AutoField(auto_created=True, primary_key=True)
-    title_key = models.CharField(max_length=255)
-    description_key = models.CharField(max_length=511)
+    title_key = models.CharField(max_length=255)  # Key used to get translated title
+    description_key = models.CharField(
+        max_length=511
+    )  # Key used to get translated description
 
 
 class Notification(models.Model):
     id = models.AutoField(auto_created=True, primary_key=True)
     user = models.ForeignKey(User, on_delete=models.CASCADE)
     template_id = models.ForeignKey(NotificationTemplate, on_delete=models.CASCADE)
     created_at = models.DateTimeField(auto_now_add=True)
-    arguments = models.JSONField(default=dict)
-    is_read = models.BooleanField(default=False)
-    is_sent = models.BooleanField(default=False)
-
-    def read(self):
-        self.is_read = True
-        self.save()
-
-    def send(self):
-        self.is_sent = True
-        self.save()
+    arguments = models.JSONField(default=dict)  # Arguments to be used in the template
+    is_read = models.BooleanField(
+        default=False
+    )  # Whether the notification has been read
+    is_sent = models.BooleanField(
+        default=False
+    )  # Whether the notification has been sent (email)

backend/notifications/serializers.py

@@ -1,5 +1,4 @@
 import re
-from os import read
 from typing import Dict, List
 
 from authentication.models import User
@@ -14,44 +13,26 @@ class Meta:
         fields = "__all__"
 
 
-class UserHyperLinkedRelatedField(serializers.HyperlinkedRelatedField):
-    view_name = "user-detail"
-    queryset = User.objects.all()
-
-    def to_internal_value(self, data):
-        try:
-            return self.queryset.get(pk=data)
-        except User.DoesNotExist:
-            self.fail("no_match")
-
-
 class NotificationSerializer(serializers.ModelSerializer):
-    user = UserHyperLinkedRelatedField()
+    # Hyper linked user field
+    user = serializers.HyperlinkedRelatedField(
+        view_name="user-detail", queryset=User.objects.all()
+    )
 
+    # Translate template and arguments into a message
     message = serializers.SerializerMethodField()
 
-    class Meta:
-        model = Notification
-        fields = [
-            "id",
-            "user",
-            "template_id",
-            "arguments",
-            "message",
-            "created_at",
-            "is_read",
-            "is_sent",
-        ]
-
-    def _get_missing_keys(self, s: str, d: Dict[str, str]) -> List[str]:
-        required_keys = re.findall(r"%\((\w+)\)", s)
-        missing_keys = [key for key in required_keys if key not in d]
+    # Check if the required arguments are present
+    def _get_missing_keys(self, string: str, arguments: Dict[str, str]) -> List[str]:
+        required_keys: List[str] = re.findall(r"%\((\w+)\)", string)
+        missing_keys = [key for key in required_keys if key not in arguments]
 
         return missing_keys
 
-    def validate(self, data):
-        data = super().validate(data)
+    def validate(self, data: Dict[str, str]) -> Dict[str, str]:
+        data: Dict[str, str] = super().validate(data)
 
+        # Validate the arguments
         if "arguments" not in data:
             data["arguments"] = {}
 
@@ -74,8 +55,22 @@ def validate(self, data):
 
         return data
 
-    def get_message(self, obj):
+    # Get the message from the template and arguments
+    def get_message(self, obj: Notification) -> Dict[str, str]:
         return {
             "title": _(obj.template_id.title_key),
             "description": _(obj.template_id.description_key) % obj.arguments,
         }
+
+    class Meta:
+        model = Notification
+        fields = [
+            "id",
+            "user",
+            "template_id",
+            "arguments",
+            "message",
+            "created_at",
+            "is_read",
+            "is_sent",
+        ]

backend/notifications/signals.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Dict
+
+from authentication.models import User
+from django.dispatch import Signal, receiver
+from django.urls import reverse
+from notifications.serializers import NotificationSerializer
+
+notification_create = Signal()
+
+
+@receiver(notification_create)
+def notification_creation(
+    type: NotificationType, user: User, arguments: Dict[str, str], **kwargs
+) -> bool:
+    serializer = NotificationSerializer(
+        data={
+            "template_id": type.value,
+            "user": reverse("user-detail", kwargs={"pk": user.id}),
+            "arguments": arguments,
+        }
+    )
+
+    if not serializer.is_valid():
+        return False
+
+    serializer.save()
+
+    return True
+
+
+class NotificationType(Enum):
+    SCORE_ADDED = 1  # Arguments: {"score": int}
+    SCORE_UPDATED = 2  # Arguments: {"score": int}

backend/notifications/urls.py

@@ -1,8 +1,6 @@
-from notifications.views import NotificationViewSet
-from rest_framework.routers import DefaultRouter
+from django.urls import path
+from notifications.views import NotificationView
 
-router = DefaultRouter()
-
-router.register(r"", NotificationViewSet, basename="notification")
-
-urlpatterns = router.urls
+urlpatterns = [
+    path("<str:user_id>/", NotificationView.as_view(), name="notification-detail"),
+]

backend/notifications/views.py

@@ -1,8 +1,38 @@
+from __future__ import annotations
+
+from typing import List
+
 from notifications.models import Notification
 from notifications.serializers import NotificationSerializer
-from rest_framework.viewsets import ModelViewSet
+from rest_framework.permissions import BasePermission, IsAuthenticated
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework.status import HTTP_200_OK
+from rest_framework.views import APIView
+
+
+# TODO: Give admin access to everything
+class NotificationPermission(BasePermission):
+    # The user can only access their own notifications
+    # An admin can access all notifications
+    def has_permission(self, request: Request, view: NotificationView) -> bool:
+        return view.kwargs.get("user_id") == request.user.id or request.user.is_staff
+
+
+class NotificationView(APIView):
+    permission_classes: List[BasePermission] = [IsAuthenticated, NotificationPermission]
+
+    def get(self, request: Request, user_id: str) -> Response:
+        notifications = Notification.objects.filter(user=user_id)
+        serializer = NotificationSerializer(
+            notifications, many=True, context={"request": request}
+        )
+
+        return Response(serializer.data)
 
+    # Mark all notifications as read for the user
+    def post(self, request: Request, user_id: str) -> Response:
+        notifications = Notification.objects.filter(user=user_id)
+        notifications.update(is_read=True)
 
-class NotificationViewSet(ModelViewSet):
-    queryset = Notification.objects.all()
-    serializer_class = NotificationSerializer
+        return Response(status=HTTP_200_OK)

backend/ypovoli/urls.py

@@ -14,18 +14,21 @@
     1. Import the include() function: from django.urls import include, path
     2. Add a URL to urlpatterns:  path('blog/', include('blog.urls'))
 """
-from django.urls import path, include
-from rest_framework import permissions
-from drf_yasg.views import get_schema_view
+
+from django.urls import include, path
 from drf_yasg import openapi
+from drf_yasg.views import get_schema_view
+from rest_framework import permissions
 
 schema_view = get_schema_view(
     openapi.Info(
         title="Ypovoli API",
         default_version="v1",
     ),
     public=True,
-    permission_classes=[permissions.AllowAny,],
+    permission_classes=[
+        permissions.AllowAny,
+    ],
 )
 
 
@@ -34,8 +37,14 @@
     path("", include("api.urls")),
     # Authentication endpoints.
     path("auth/", include("authentication.urls")),
-    path("notifications/", include("notifications.urls")),
+    path("notifications/", include("notifications.urls"), name="notifications"),
     # Swagger documentation.
-    path("swagger/", schema_view.with_ui("swagger", cache_timeout=0), name="schema-swagger-ui"),
-    path("swagger<format>/", schema_view.without_ui(cache_timeout=0), name="schema-json"),
+    path(
+        "swagger/",
+        schema_view.with_ui("swagger", cache_timeout=0),
+        name="schema-swagger-ui",
+    ),
+    path(
+        "swagger<format>/", schema_view.without_ui(cache_timeout=0), name="schema-json"
+    ),
 ]