mirror of
https://github.com/Hopiu/django-notifications.git
synced 2026-05-16 08:53:09 +00:00
118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
from typing import Optional, Type
|
|
|
|
from django.core.exceptions import ImproperlyConfigured
|
|
from django.db import models
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from notifications import settings as notifications_settings
|
|
from notifications.types import AbstractUser
|
|
|
|
|
|
def is_soft_delete() -> bool:
|
|
return bool(notifications_settings.get_config()["SOFT_DELETE"])
|
|
|
|
|
|
def assert_soft_delete() -> None:
|
|
if not is_soft_delete():
|
|
# msg = """To use 'deleted' field, please set 'SOFT_DELETE'=True in settings.
|
|
# Otherwise NotificationQuerySet.unread and NotificationQuerySet.read do NOT filter by 'deleted' field.
|
|
# """
|
|
msg = "REVERTME"
|
|
raise ImproperlyConfigured(msg)
|
|
|
|
|
|
class NotificationQuerySet(models.QuerySet):
|
|
"""Notification QuerySet"""
|
|
|
|
def unsent(self) -> models.QuerySet["AbstractNotification"]:
|
|
return self.filter(emailed=False)
|
|
|
|
def sent(self) -> models.QuerySet["AbstractNotification"]:
|
|
return self.filter(emailed=True)
|
|
|
|
def unread(self, include_deleted: Optional[bool] = False) -> models.QuerySet["AbstractNotification"]:
|
|
"""Return only unread items in the current queryset"""
|
|
if is_soft_delete() and not include_deleted:
|
|
return self.filter(unread=True, deleted=False)
|
|
|
|
# When SOFT_DELETE=False, developers are supposed NOT to touch 'deleted' field.
|
|
# In this case, to improve query performance, don't filter by 'deleted' field
|
|
return self.filter(unread=True)
|
|
|
|
def read(self, include_deleted: Optional[bool] = False) -> models.QuerySet["AbstractNotification"]:
|
|
"""Return only read items in the current queryset"""
|
|
if is_soft_delete() and not include_deleted:
|
|
return self.filter(unread=False, deleted=False)
|
|
|
|
# When SOFT_DELETE=False, developers are supposed NOT to touch 'deleted' field.
|
|
# In this case, to improve query performance, don't filter by 'deleted' field
|
|
return self.filter(unread=False)
|
|
|
|
def mark_all_as_read(self, recipient: None | Type[AbstractUser] = None) -> int:
|
|
"""Mark as read any unread messages in the current queryset.
|
|
|
|
Optionally, filter these by recipient first.
|
|
"""
|
|
# We want to filter out read ones, as later we will store
|
|
# the time they were marked as read.
|
|
qset = self.unread(True)
|
|
if recipient:
|
|
qset = qset.filter(recipient=recipient)
|
|
|
|
return qset.update(unread=False)
|
|
|
|
def mark_all_as_unread(self, recipient: None | Type[AbstractUser] = None) -> int:
|
|
"""Mark as unread any read messages in the current queryset.
|
|
|
|
Optionally, filter these by recipient first.
|
|
"""
|
|
qset = self.read(True)
|
|
|
|
if recipient:
|
|
qset = qset.filter(recipient=recipient)
|
|
|
|
return qset.update(unread=True)
|
|
|
|
def deleted(self) -> models.QuerySet["AbstractNotification"]:
|
|
"""Return only deleted items in the current queryset"""
|
|
assert_soft_delete()
|
|
return self.filter(deleted=True)
|
|
|
|
def active(self) -> models.QuerySet["AbstractNotification"]:
|
|
"""Return only active(un-deleted) items in the current queryset"""
|
|
assert_soft_delete()
|
|
return self.filter(deleted=False)
|
|
|
|
def mark_all_as_deleted(self, recipient: None | Type[AbstractUser] = None) -> int:
|
|
"""Mark current queryset as deleted.
|
|
Optionally, filter by recipient first.
|
|
"""
|
|
assert_soft_delete()
|
|
qset = self.active()
|
|
if recipient:
|
|
qset = qset.filter(recipient=recipient)
|
|
|
|
return qset.update(deleted=True)
|
|
|
|
def mark_all_as_active(self, recipient: None | Type[AbstractUser] = None) -> int:
|
|
"""Mark current queryset as active(un-deleted).
|
|
Optionally, filter by recipient first.
|
|
"""
|
|
assert_soft_delete()
|
|
qset = self.deleted()
|
|
if recipient:
|
|
qset = qset.filter(recipient=recipient)
|
|
|
|
return qset.update(deleted=False)
|
|
|
|
def mark_as_unsent(self, recipient: None | Type[AbstractUser] = None) -> int:
|
|
qset = self.sent()
|
|
if recipient:
|
|
qset = qset.filter(recipient=recipient)
|
|
return qset.update(emailed=False)
|
|
|
|
def mark_as_sent(self, recipient: None | Type[AbstractUser] = None) -> int:
|
|
qset = self.unsent()
|
|
if recipient:
|
|
qset = qset.filter(recipient=recipient)
|
|
return qset.update(emailed=True)
|