Coverage for baibu/notifications/services.py: 100%

1"""Sending and reading notifications. 

2 

3Feature code calls :func:`notify`; it never fails the caller's work. Use a 

4kind registered in :mod:`.kinds` so the text is shown in the reader's 

5language; ``title`` and ``body`` are a fallback for unregistered kinds. 

6""" 

7 

8import logging 

9from datetime import timedelta 

10 

11from django.conf import settings 

12from django.utils import timezone 

13 

14from .models import Notification 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19def notify( # noqa: PLR0913 

20 user, *, kind: str, params: dict | None = None, title: str = "", body: str = "", link: str = "" 

21) -> Notification | None: 

22 if not settings.NOTIFICATIONS_ENABLED or not user.is_active: 

23 return None 

24 if link and not link.startswith("/"): 

25 # Only links inside the platform. 

26 link = "" 

27 try: 

28 return Notification.objects.create( 

29 user=user, 

30 kind=kind[:64], 

31 params=params or {}, 

32 title=str(title)[:200], 

33 body=str(body), 

34 link=link[:500], 

35 ) 

36 except Exception: 

37 logger.exception("Could not create a %s notification", kind) 

38 return None 

39 

40 

41def unread_count(user) -> int: 

42 return Notification.objects.filter(user=user, read_at__isnull=True).count() 

43 

44 

45def mark_read(notification: Notification) -> None: 

46 if notification.read_at is None: 

47 notification.read_at = timezone.now() 

48 notification.save(update_fields=["read_at"]) 

49 

50 

51def mark_all_read(user) -> int: 

52 return Notification.objects.filter(user=user, read_at__isnull=True).update(read_at=timezone.now()) 

53 

54 

55def delete_old(now=None) -> int: 

56 """Delete read notifications older than the retention period.""" 

57 now = now or timezone.now() 

58 cutoff = now - timedelta(days=settings.NOTIFICATIONS_RETENTION_DAYS) 

59 deleted, _ = Notification.objects.filter(read_at__isnull=False, created_at__lt=cutoff).delete() 

60 return deleted