Coverage for modules\notifications\api.py: 58%

1from ninja import Router 

2from typing import List, Optional 

3from uuid import UUID 

4from core.auth import JWTAuth 

5from core.permissions import require_tenant 

6from .schemas import NotificationOut 

7from .models import Notification 

8from .services import NotificationService 

9 

10router = Router(tags=["Notifications"]) 

11 

12@router.get("", auth=JWTAuth(), response={200: List[NotificationOut]}) 

13def list_notifications(request, unread_only: bool = False): 

14 tenant_id = require_tenant(request) 

15 qs = Notification.objects.filter(recipient=request.user, tenant_id=tenant_id) 

16 if unread_only: 

17 qs = qs.filter(is_read=False) 

18 return list(qs[:50]) 

19 

20@router.post("/read-all", auth=JWTAuth(), response={200: dict}) 

21def mark_all_read(request): 

22 tenant_id = require_tenant(request) 

23 NotificationService.mark_as_read(tenant_id, request.user) 

24 return {"success": True} 

25 

26@router.post("/{notification_id}/read", auth=JWTAuth(), response={200: dict}) 

27def mark_read(request, notification_id: UUID): 

28 tenant_id = require_tenant(request) 

29 NotificationService.mark_as_read(tenant_id, request.user, notification_id) 

30 return {"success": True}