sajad-dev/Backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

UPDATE

Browse Source
This commit is contained in:
Mohammad Sajad Pourajam
2026-04-05 05:10:17 +03:30
parent ec8e72737b
commit 6f098a1021
6 changed files with 457 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
notifications/NOTIFICATION_API_CHANGES.md
+80
View File
@@ -0,0 +1,80 @@
# Notification API Changes
## Added paginated notification list API
A new endpoint was added to return all notifications for a farm using pagination.
### Endpoint
`GET /api/notifications/list/`
### Query params
- `farm_uuid` (required): UUID of the farm
- `page` (optional): page number, default depends on DRF pagination behavior
- `page_size` (optional): number of items per page, default `10`, max `100`
### Behavior
- Requires authenticated user
- Returns notifications only if the farm belongs to the authenticated user
- Orders notifications by newest first using `created_at DESC, id DESC`
- Returns paginated response
- Returns `404` if the farm is not found or does not belong to the user
### Response shape
```json
{
"count": 12,
"next": "http://localhost:8000/api/notifications/list/?farm_uuid=<uuid>&page=2&page_size=5",
"previous": null,
"results": {
"code": 200,
"msg": "success",
"data": [
{
"uuid": "...",
"farm_uuid": "...",
"since_id": 12,
"title": "Alert",
"message": "Check sensor",
"level": "info",
"is_read": false,
"metadata": {},
"created_at": "2025-01-01T10:00:00Z"
}
]
}
}
```
## Long-poll behavior update
The `long-poll` notification logic was updated to prioritize unread notifications.
### Updated behavior
- Returns unread notifications first
- If unread notifications are fewer than `5`, fills the remaining slots with read notifications
- If unread notifications are `5` or more, returns only the first `5` unread notifications
### Notes
This behavior was implemented in the notification service layer so the existing long-poll endpoint automatically uses it.
## Files changed
- `notifications/views.py`
- `notifications/urls.py`
- `notifications/services.py`
- `notifications/tests.py`
## Tests added
Added tests for:
- paginated notification list for owned farm
- `404` for unowned farm on list API
- unread-first ordering in long-poll
- long-poll fallback with read notifications when unread count is below `5`
notifications/services.py
+28 -3
View File
@@ -1,7 +1,8 @@
import time
from django.db import OperationalError, ProgrammingError
from django.db.models import QuerySet
from django.db.models import Case, IntegerField, QuerySet, Value, When
from farm_hub.models import FarmHub
@@ -39,6 +40,30 @@ def get_notifications_for_farm(*, farm: FarmHub, since_id=None) -> QuerySet[Farm
raise ValueError("Notifications table is not migrated.") from exc
def get_prioritized_notifications_for_farm(*, farm: FarmHub, since_id=None, limit=5) -> QuerySet[FarmNotification]:
try:
unread_queryset = get_notifications_for_farm(farm=farm, since_id=since_id).filter(is_read=False)
unread_count = unread_queryset.count()
if unread_count >= limit:
return unread_queryset[:limit]
fallback_limit = max(limit - unread_count, 0)
if fallback_limit == 0:
return unread_queryset
queryset = get_notifications_for_farm(farm=farm, since_id=since_id).annotate(
priority=Case(
When(is_read=False, then=Value(0)),
default=Value(1),
output_field=IntegerField(),
)
)
return queryset.order_by("priority", "created_at", "id")[:limit]
except (ProgrammingError, OperationalError) as exc:
raise ValueError("Notifications table is not migrated.") from exc
def mark_notifications_as_read(*, farm: FarmHub, slice_id: int) -> int:
try:
return FarmNotification.objects.filter(
@@ -50,10 +75,10 @@ def mark_notifications_as_read(*, farm: FarmHub, slice_id: int) -> int:
raise ValueError("Notifications table is not migrated.") from exc
def long_poll_notifications(*, farm: FarmHub, since_id=None, timeout_seconds=DEFAULT_POLL_TIMEOUT_SECONDS, interval_seconds=DEFAULT_POLL_INTERVAL_SECONDS):
def long_poll_notifications(*, farm: FarmHub, since_id=None, timeout_seconds=DEFAULT_POLL_TIMEOUT_SECONDS, interval_seconds=DEFAULT_POLL_INTERVAL_SECONDS, limit=5):
deadline = time.monotonic() + max(timeout_seconds, 0)
while True:
notifications = list(get_notifications_for_farm(farm=farm, since_id=since_id))
notifications = list(get_prioritized_notifications_for_farm(farm=farm, since_id=since_id, limit=limit))
if notifications:
return notifications
if time.monotonic() >= deadline:
notifications/tests.py
+87 -2
View File
@@ -8,8 +8,8 @@ from rest_framework.test import APIRequestFactory, force_authenticate
from farm_hub.models import FarmHub, FarmType
from .models import FarmNotification
from .services import create_notification_for_farm_uuid, long_poll_notifications, mark_notifications_as_read
from .views import ExternalNotificationIngestView, NotificationLongPollView, NotificationMarkReadView
from .services import create_notification_for_farm_uuid, get_prioritized_notifications_for_farm, long_poll_notifications, mark_notifications_as_read
from .views import ExternalNotificationIngestView, NotificationListView, NotificationLongPollView, NotificationMarkReadView
class NotificationServiceTests(TestCase):
@@ -72,6 +72,26 @@ class NotificationServiceTests(TestCase):
self.assertFalse(third.is_read)
def test_get_prioritized_notifications_for_farm_returns_unread_first_and_fills_with_read(self):
unread_one = FarmNotification.objects.create(farm=self.farm, title="Unread 1", message="A")
unread_two = FarmNotification.objects.create(farm=self.farm, title="Unread 2", message="B")
read_one = FarmNotification.objects.create(farm=self.farm, title="Read 1", message="C", is_read=True)
read_two = FarmNotification.objects.create(farm=self.farm, title="Read 2", message="D", is_read=True)
notifications = list(get_prioritized_notifications_for_farm(farm=self.farm, limit=5))
self.assertEqual([notification.id for notification in notifications], [unread_one.id, unread_two.id, read_one.id, read_two.id])
def test_long_poll_notifications_limits_to_five_unread_notifications(self):
for index in range(6):
FarmNotification.objects.create(farm=self.farm, title=f"Unread {index}", message="B")
notifications = long_poll_notifications(farm=self.farm, timeout_seconds=0, limit=5)
self.assertEqual(len(notifications), 5)
self.assertTrue(all(notification.is_read is False for notification in notifications))
class NotificationLongPollViewTests(TestCase):
def setUp(self):
self.factory = APIRequestFactory()
@@ -107,6 +127,23 @@ class NotificationLongPollViewTests(TestCase):
self.assertEqual(response.data["data"][0]["since_id"], notification.id)
self.assertFalse(response.data["data"][0]["is_read"])
def test_long_poll_view_returns_unread_first_then_read_when_unread_is_less_than_five(self):
unread_one = FarmNotification.objects.create(farm=self.farm, title="Unread 1", message="A")
unread_two = FarmNotification.objects.create(farm=self.farm, title="Unread 2", message="B")
read_one = FarmNotification.objects.create(farm=self.farm, title="Read 1", message="C", is_read=True)
read_two = FarmNotification.objects.create(farm=self.farm, title="Read 2", message="D", is_read=True)
request = self.factory.get(f"/api/notifications/long-poll/?farm_uuid={self.farm.farm_uuid}&timeout=0")
force_authenticate(request, user=self.user)
response = NotificationLongPollView.as_view()(request)
self.assertEqual(response.status_code, 200)
self.assertEqual(
[item["since_id"] for item in response.data["data"]],
[unread_one.id, unread_two.id, read_one.id, read_two.id],
)
self.assertEqual([item["is_read"] for item in response.data["data"]], [False, False, True, True])
def test_long_poll_view_returns_404_for_unowned_farm(self):
request = self.factory.get(f"/api/notifications/long-poll/?farm_uuid={self.farm.farm_uuid}&timeout=0")
force_authenticate(request, user=self.other_user)
@@ -196,6 +233,54 @@ class NotificationMarkReadViewTests(TestCase):
self.assertEqual(response.data["msg"], "Farm not found.")
class NotificationListViewTests(TestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.user = get_user_model().objects.create_user(
username="notif-list-user",
password="secret123",
email="notif-list@example.com",
phone_number="09120000017",
)
self.other_user = get_user_model().objects.create_user(
username="notif-list-other-user",
password="secret123",
email="notif-list-other@example.com",
phone_number="09120000018",
)
self.farm_type = FarmType.objects.create(name="مرغداری")
self.farm = FarmHub.objects.create(
owner=self.user,
farm_type=self.farm_type,
name="Farm E",
)
def test_list_view_returns_paginated_notifications_for_owned_farm(self):
for index in range(12):
FarmNotification.objects.create(farm=self.farm, title=f"Alert {index}", message="Check sensor")
request = self.factory.get(
f"/api/notifications/list/?farm_uuid={self.farm.farm_uuid}&page=2&page_size=5"
)
force_authenticate(request, user=self.user)
response = NotificationListView.as_view()(request)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["count"], 12)
self.assertEqual(len(response.data["results"]["data"]), 5)
self.assertEqual(response.data["results"]["code"], 200)
def test_list_view_returns_404_for_unowned_farm(self):
request = self.factory.get(f"/api/notifications/list/?farm_uuid={self.farm.farm_uuid}")
force_authenticate(request, user=self.other_user)
response = NotificationListView.as_view()(request)
self.assertEqual(response.status_code, 404)
self.assertEqual(response.data["msg"], "Farm not found.")
@override_settings(EXTERNAL_NOTIFICATION_API_KEY="12345")
class ExternalNotificationIngestViewTests(TestCase):
def setUp(self):
notifications/urls.py
+2 -1
View File
@@ -1,8 +1,9 @@
from django.urls import path
from .views import NotificationLongPollView, NotificationMarkReadView
from .views import NotificationListView, NotificationLongPollView, NotificationMarkReadView
urlpatterns = [
path("list/", NotificationListView.as_view(), name="notification-list"),
path("long-poll/", NotificationLongPollView.as_view(), name="notification-long-poll"),
path("mark-as-read/", NotificationMarkReadView.as_view(), name="notification-mark-as-read"),
]
notifications/views.py
+46
View File
@@ -1,6 +1,7 @@
from django.conf import settings
from drf_spectacular.utils import OpenApiParameter, OpenApiTypes, extend_schema
from rest_framework import serializers, status
from rest_framework.pagination import PageNumberPagination
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
@@ -18,6 +19,16 @@ class NotificationLongPollQuerySerializer(serializers.Serializer):
timeout = serializers.IntegerField(required=False, min_value=0, max_value=60)
class NotificationListQuerySerializer(serializers.Serializer):
farm_uuid = serializers.UUIDField()
page = serializers.IntegerField(required=False, min_value=1)
page_size = serializers.IntegerField(required=False, min_value=1, max_value=100)
class NotificationPagination(PageNumberPagination):
page_size = 10
page_size_query_param = "page_size"
max_page_size = 100
class NotificationMarkReadSerializer(serializers.Serializer):
@@ -69,6 +80,41 @@ class NotificationLongPollView(APIView):
return Response({"code": 200, "msg": "success", "data": data}, status=status.HTTP_200_OK)
class NotificationListView(APIView):
permission_classes = [IsAuthenticated]
pagination_class = NotificationPagination
@extend_schema(
tags=["Notifications"],
parameters=[NotificationListQuerySerializer],
responses={
200: code_response("NotificationListResponse"),
404: code_response("NotificationListNotFoundResponse"),
},
)
def get(self, request):
serializer = NotificationListQuerySerializer(data=request.query_params)
serializer.is_valid(raise_exception=True)
farm = get_owned_farm(
farm_uuid=serializer.validated_data["farm_uuid"],
user=request.user,
)
if farm is None:
return Response({"code": 404, "msg": "Farm not found."}, status=status.HTTP_404_NOT_FOUND)
paginator = self.pagination_class()
notifications = farm.notifications.all().order_by("-created_at", "-id")
page = paginator.paginate_queryset(notifications, request, view=self)
data = FarmNotificationSerializer(page, many=True).data
return paginator.get_paginated_response({
"code": 200,
"msg": "success",
"data": data,
})
class NotificationMarkReadView(APIView):
permission_classes = [IsAuthenticated]