Push Notifications API

src/api/urls.py

@@ -0,0 +1,12 @@
+from django.urls import include, path
+from drf_spectacular import views as drf_views
+
+from . import v1
+
+app_name = "api"
+urlpatterns = [
+    path("v1/", include((v1.urls, "api"), namespace="v1")),
+    path("schema/", drf_views.SpectacularAPIView.as_view(), name="schema"),
+    path("docs/", drf_views.SpectacularSwaggerView.as_view(url_name="schema"), name="docs"),
+    path("redoc/", drf_views.SpectacularRedocView.as_view(url_name="schema"), name="redoc"),
+]

src/api/v1/__init__.py

@@ -0,0 +1 @@
+from . import urls

src/api/v1/push_notifications/__init__.py

@@ -0,0 +1 @@
+from . import urls

src/api/v1/push_notifications/schemas.py

@@ -0,0 +1,53 @@
+from rest_framework import serializers
+
+from app.consts import push_notification as push_notification_consts
+from app.drf.fields import create_choice_human_field
+from app.drf.serializers import inline_serializer
+from push_notifications import models
+
+
+class PushNotificationInputSchema(serializers.Serializer):
+    only_read = serializers.BooleanField(required=False, allow_null=True)
+
+
+PushNotificationKindField = create_choice_human_field(constant_class=push_notification_consts.Kind)
+PushNotificationStatusField = create_choice_human_field(
+    constant_class=push_notification_consts.Status
+)
+
+
+class PushNotificationOutputSchema(serializers.ModelSerializer):
+    kind = PushNotificationKindField()
+    status = PushNotificationStatusField()
+    data = inline_serializer(
+        name="PushNotificationEnrichedDataOutputSchema",
+        fields={
+            "id": serializers.CharField(),
+            "createdAt": serializers.DateTimeField(),
+            "readAt": serializers.DateTimeField(),
+            "timeSinceCreated": serializers.CharField(),
+            "kind": serializers.CharField(),
+            "meta": serializers.JSONField(),
+        },
+        source="enriched_data",
+    )
+
+    class Meta:
+        model = models.PushNotification
+        fields = (
+            "id",
+            "title",
+            "description",
+            "read_at",
+            "kind",
+            "status",
+            "data",
+        )
+
+
+class PushNotificationReadManyInputSchema(serializers.Serializer):
+    ids = serializers.ListField(required=False, allow_null=True)
+
+
+class PushNotificationReadManyOutputSchema(serializers.Serializer):
+    read = serializers.IntegerField()

src/api/v1/push_notifications/urls.py

@@ -0,0 +1,8 @@
+from rest_framework.routers import SimpleRouter
+
+from . import views
+
+router = SimpleRouter(trailing_slash=False)
+router.register("notifications", views.PushNotificationViewSet, basename="push_notifications")
+
+urlpatterns = router.urls

src/api/v1/push_notifications/views.py

@@ -0,0 +1,69 @@
+from rest_framework.decorators import action
+from rest_framework.generics import get_object_or_404
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from app.drf.openapi import limit_offset_openapi_schema, openapi_schema
+from app.drf.viewsets import AppViewSet
+from push_notifications import selectors, services
+
+from . import schemas
+
+
+class PushNotificationViewSet(AppViewSet):
+    permission_classes = [IsAuthenticated]
+
+    @limit_offset_openapi_schema(
+        wrapped_schema=schemas.PushNotificationOutputSchema,
+        operation_id="push-notification-list",
+        summary="Notifications list",
+        description="Returns a list of notifications",
+        request=None,
+        tags=["push notifications"],
+        add_unauthorized_response=True,
+        parameter_serializer=schemas.PushNotificationInputSchema,
+    )
+    def list(self, request: Request) -> Response:
+        params = self.get_valid_query_params(srlzr_class=schemas.PushNotificationInputSchema)
+        qs = selectors.push_notification_get_viewable_qs(user=request.user, filters=params)
+        return self.get_paginated_response(
+            queryset=qs, srlzr_class=schemas.PushNotificationOutputSchema
+        )
+
+    @openapi_schema(
+        summary="Read many notifications",
+        description="Reads a list of notifications",
+        request=schemas.PushNotificationReadManyInputSchema,
+        responses={200: schemas.PushNotificationReadManyOutputSchema},
+        tags=["push notifications"],
+        operation_id="push-notifications-read",
+        add_bad_request_response=True,
+        add_unauthorized_response=True,
+    )
+    @action(methods=["PATCH"], detail=False, url_path="read")
+    def read_many(self, request: Request) -> Response:
+        data = self.get_valid_data(srlzr_class=schemas.PushNotificationReadManyInputSchema)
+        updated = services.push_notification_read_many(reader=request.user, ids=data.get("ids"))
+        return Response(data={"read": updated})
+
+    @openapi_schema(
+        summary="Read notification",
+        description="Reads a single notification",
+        request=None,
+        responses={200: schemas.PushNotificationOutputSchema},
+        tags=["push notifications"],
+        operation_id="push-notification-read",
+        add_not_found_response=True,
+        add_bad_request_response=True,
+        add_unauthorized_response=True,
+    )
+    @action(methods=["PATCH"], detail=True)
+    def read(self, request: Request, id: int) -> Response:
+        notification = get_object_or_404(
+            selectors.push_notification_get_viewable_qs(user=request.user),
+            pk=id,
+        )
+        notification = services.push_notification_read(push_notification=notification)
+        out_srlzr = schemas.PushNotificationOutputSchema(instance=notification)
+        return Response(data=out_srlzr.data)

src/api/v1/urls.py

@@ -0,0 +1,10 @@
+from django.urls import include, path
+
+from . import push_notifications
+
+urlpatterns = [
+    path(
+        "notifications/",
+        include((push_notifications.urls, "push_notifications"), namespace="push_notifications"),
+    ),
+]

src/app/drf/fields.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from drf_spectacular.utils import extend_schema_field
 from rest_framework import serializers
 
@@ -19,12 +21,20 @@ def __call__(self, serializer_field):
         return serializer_field.context[self.context_key]
 
 
-def create_choice_human_field(name: str, choices, **kwargs) -> type[serializers.ChoiceField]:
+def create_choice_human_field(
+    constant_class: type[Any],
+    choices_attr: str = "choices",
+) -> type[serializers.ChoiceField]:
     """Function that wraps the creation of a choice field.
     This is a function and not a class directly in order to
     define the choices inside the `extend_schema_field`
     this will populate the documentation correctly"""
 
+    class_name = constant_class.__name__
+    choices = getattr(constant_class, choices_attr, None)
+    if choices is None:
+        raise TypeError(f"Missing {choices_attr=} on {class_name=}")
+
     class BaseHumanField(serializers.ChoiceField):
         def __init__(self, **kwargs):
             super().__init__(choices, **kwargs)
@@ -34,7 +44,7 @@ def to_representation(self, value):
                 return None
             return {"value": value, "human": self._choices.get(value, None)}
 
-    field_name = "{}ChoiceHumanField".format(name.title())
+    field_name = "{}ChoiceHumanField".format(class_name.title())
     _ChoiceHumanField = type(field_name, (BaseHumanField,), {})
     return extend_schema_field(
         inline_serializer(

src/app/drf/utils.py

@@ -0,0 +1,19 @@
+from typing import Any
+
+from rest_framework import request, serializers
+
+
+def get_valid_request_query_params(
+    *, request: request.Request, srlzr_class: type[serializers.Serializer]
+) -> dict[str, Any]:
+    srlzr = srlzr_class(data=request.query_params)
+    srlzr.is_valid(raise_exception=True)
+    return srlzr.validated_data
+
+
+def get_valid_request_data(
+    *, request: request.Request, srlzr_class: type[serializers.Serializer]
+) -> dict[str, Any]:
+    srlzr = srlzr_class(data=request.data)
+    srlzr.is_valid(raise_exception=True)
+    return srlzr.validated_data

src/app/drf/viewsets.py

@@ -0,0 +1,30 @@
+from typing import Any
+
+from django.db.models import QuerySet
+from rest_framework import serializers, viewsets
+from rest_framework.pagination import BasePagination
+
+from . import pagination, utils
+
+
+class AppViewSet(viewsets.ViewSet):
+    def get_valid_query_params(self, srlzr_class: type[serializers.Serializer]):
+        return utils.get_valid_request_query_params(request=self.request, srlzr_class=srlzr_class)
+
+    def get_valid_data(self, srlzr_class: type[serializers.Serializer]):
+        return utils.get_valid_request_data(request=self.request, srlzr_class=srlzr_class)
+
+    def get_paginated_response(
+        self,
+        queryset: QuerySet,
+        srlzr_class: type[serializers.Serializer],
+        srlzr_context: dict[str, Any] | None = None,
+        pagination_class: type[BasePagination] | None = None,
+    ):
+        return pagination.get_paginated_response(
+            view=self,
+            queryset=queryset,
+            serializer_class=srlzr_class,
+            extra_context=srlzr_context,
+            pagination_class=pagination_class or pagination.LimitOffsetPagination,
+        )

src/app/settings/conf.py

@@ -1,6 +1,6 @@
 import os
 from pathlib import Path
-from typing import Any, Dict, List, Tuple
+from typing import Any, Dict, List
 
 import environ
 from django.utils.translation import gettext_lazy as _

src/app/urls.py

@@ -40,4 +40,5 @@
     path("api/schema/", auth_decorators.login_required(schema_view), name="schema"),
     path("api/docs/", auth_decorators.login_required(swagger_view), name="docs"),
     path("api/redoc/", auth_decorators.login_required(redoc_view), name="redoc"),
+    path("api/", include("api.urls", namespace="api")),
 ]

src/push_notifications/services.py

@@ -14,7 +14,7 @@
 from app.models import BaseModel
 from users.models import User
 
-from . import models
+from . import models, selectors
 
 
 def push_notification_create(
@@ -31,8 +31,8 @@ def push_notification_create(
     push_notification = models.PushNotification(
         user=user,
         kind=kind,
-        title=title,
-        description=description,
+        title=title[: consts.push_notification.MaxSize.TITLE_MAX_SIZE_ANDROID],
+        description=description[: consts.push_notification.MaxSize.DESCRIPTION_MAX_SIZE_ANDROID],
         data=data,
         status=consts.push_notification.Status.CREATED,
         source_object=source_object,
@@ -229,3 +229,13 @@ def push_notification_read(
     push_notification.status = consts.push_notification.Status.READ
     push_notification.save()
     return push_notification
+
+
+def push_notification_read_many(*, reader: User, ids: Sequence[int]) -> int:
+    """Reads a sequence of ids visible for the given `reader`. If `ids` is empty
+    reads all unread notifications.
+    Returns the updated notification count"""
+    qs = selectors.push_notification_get_viewable_qs(user=reader).filter(read_at__isnull=True)
+    if ids:
+        qs = qs.filter(pk__in=ids)
+    return qs.update(read_at=timezone.now(), status=consts.push_notification.Status.READ)

src/push_notifications/tasks.py

@@ -5,6 +5,12 @@
 from . import models, services
 
 
+@task()
+def push_notification_send(notification_ids: list[int]):
+    notifications = models.PushNotification.objects.filter(id__in=notification_ids)
+    services.push_notification_send(notifications=notifications)
+
+
 @task()
 def push_notification_handle_delivery_failure(notification_id: int):
     notification = models.PushNotification.objects.get(pk=notification_id)