SystemNotifications handling in Django REST

Published On: 17 December 2020.By Auriga User.

General

SystemNotifications handling in Django REST

To manage notifications we define types of notifications in a constant files(in a dictionary) and create two tables in our database, first is NotificationObjects and second, Notifications like:

class NotificationObjects(models.Model):
    class Meta:
        verbose_name_plural = "Notification Objects"

    def __str__(self):
        return str(self.id)

    notices = NOTIFICATION_ENTITY_TYPE_CHOICES

    entity_keys = list(notices.keys())
    entity_values = list(notices.values())
    notice_tuples_list = []

    for i in range(len(entity_keys)):
        notice_tuples_list.append((entity_values[i],entity_keys[i]))

    NOTIFICATION_OBJECT_ENTITY_TYPE_CHOICES = notice_tuples_list



    entity_type_id = models.IntegerField(default=0, blank=True, choices=NOTIFICATION_OBJECT_ENTITY_TYPE_CHOICES)
    entity_id = models.IntegerField(default=0, blank=True)
    actor = models.ForeignKey(InfaUsers, on_delete=models.CASCADE)
    is_sent = models.BooleanField(default=False)

    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    deleted_at = models.DateTimeField(auto_now=True)

class NotificationObjects(models.Model):

class Meta:

verbose_name_plural = "Notification Objects"

def __str__(self):

return str(self.id)

notices = NOTIFICATION_ENTITY_TYPE_CHOICES

entity_keys = list(notices.keys())

entity_values = list(notices.values())

notice_tuples_list = []

for i in range(len(entity_keys)):

notice_tuples_list.append((entity_values[i],entity_keys[i]))

NOTIFICATION_OBJECT_ENTITY_TYPE_CHOICES = notice_tuples_list

entity_type_id = models.IntegerField(default=0, blank=True, choices=NOTIFICATION_OBJECT_ENTITY_TYPE_CHOICES)

entity_id = models.IntegerField(default=0, blank=True)

actor = models.ForeignKey(InfaUsers, on_delete=models.CASCADE)

is_sent = models.BooleanField(default=False)

is_active = models.BooleanField(default=True)

created_at = models.DateTimeField(auto_now_add=True)

updated_at = models.DateTimeField(auto_now=True)

deleted_at = models.DateTimeField(auto_now=True)

class Notifications(models.Model):
    class Meta:
        verbose_name_plural = "Notifications"

    def __str__(self):
        return str(self.id)
    notification_object = models.ForeignKey(NotificationObjects, on_delete=models.CASCADE)
    notifier = models.ForeignKey(InfaUsers, on_delete=models.CASCADE)
    is_sent = models.BooleanField(default=False)
    is_read = models.BooleanField(default=False)

    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    deleted_at = models.DateTimeField(auto_now=True)

class Notifications(models.Model):

class Meta:

verbose_name_plural = "Notifications"

def __str__(self):

return str(self.id)

notification_object = models.ForeignKey(NotificationObjects, on_delete=models.CASCADE)

notifier = models.ForeignKey(InfaUsers, on_delete=models.CASCADE)

is_sent = models.BooleanField(default=False)

is_read = models.BooleanField(default=False)

is_active = models.BooleanField(default=True)

created_at = models.DateTimeField(auto_now_add=True)

updated_at = models.DateTimeField(auto_now=True)

deleted_at = models.DateTimeField(auto_now=True)

NOTIFICATION_ENTITY_TYPE_CHOICES = {
    "File_uplaod": 1,
    "New_message": 2,...}

NOTIFICATION_ENTITY_TYPE_CHOICES = {

"File_uplaod": 1,

"New_message": 2,...}

In NotificationObjects table we define entity_type_id which keeps the type of notification we defined in constant file (e.g. {“file_upload_notification”: 1,…}) and a entity_id which keeps id of related thing; like for file_upload type of notification we keep file id in entity_id to fetch details of of file from file table through its id. Next in actor field we put id of user who is creating the notification.

In next table i.e. Notifications we put foreign key which references notificationobject, notifier which is the user to whom notification to be sent, is_read to mark if notification is read by the notified user.

Next we create two functions to create notificationObject and notification.

def create_notification_object(entity_type,entity_id,actor, is_sent=True):
    try:
        entity_type_id = NOTIFICATION_ENTITY_TYPE_CHOICES.get(entity_type)

        notification_object_data = {
            'entity_type_id': entity_type_id,
            'entity_id': entity_id,
            'actor':actor,
            'is_sent': is_sent,

        }

        notification_object_serializer = NotificationObjectsSerializer(data=notification_object_data)

        if notification_object_serializer.is_valid():
            notification_object_serializer.save()
            notification_object_id=notification_object_serializer.data.get('id')
            return True, notification_object_id
        else:
            responseData = {"message": "Invalid request.", "status":False, "status_code":status.HTTP_300_MULTIPLE_CHOICES ,"dev_message": "invalid case id"}
            return False, responseData

    except Exception as e:
        log.error(str(e))
        return False, {"message": "Exception", "dev_message": str(e), "status_code": status.HTTP_300_MULTIPLE_CHOICES, "status": False}

def create_notification_object(entity_type,entity_id,actor, is_sent=True):

try:

entity_type_id = NOTIFICATION_ENTITY_TYPE_CHOICES.get(entity_type)

notification_object_data = {

'entity_type_id': entity_type_id,

'entity_id': entity_id,

'actor':actor,

'is_sent': is_sent,

}

notification_object_serializer = NotificationObjectsSerializer(data=notification_object_data)

if notification_object_serializer.is_valid():

notification_object_serializer.save()

notification_object_id=notification_object_serializer.data.get('id')

return True, notification_object_id

else:

responseData = {"message": "Invalid request.", "status":False, "status_code":status.HTTP_300_MULTIPLE_CHOICES ,"dev_message": "invalid case id"}

return False, responseData

except Exception as e:

log.error(str(e))

return False, {"message": "Exception", "dev_message": str(e), "status_code": status.HTTP_300_MULTIPLE_CHOICES, "status": False}

def create_notification(notification_object_id,notifier,is_sent=True):

    try:

        notification_data={
            'notification_object':notification_object_id,
            'notifier':notifier,
            'is_sent': is_sent
        }

        notification_serializer = NotificationsSerializer(data=notification_data)

        if notification_serializer.is_valid():
            notification_serializer.save()
            return True, {"notification created"}
        else:
            responseData = {"message": "Invalid request.", "status":False, "status_code":status.HTTP_300_MULTIPLE_CHOICES ,"dev_message": "invalid case id"}
            return False, responseData

    except Exception as e:
        log.error("Exception in create_notification: " + str(e))
        return False, {"message": "Exception", "dev_message": str(e), "status_code": status.HTTP_300_MULTIPLE_CHOICES, "status": False}

def create_notification(notification_object_id,notifier,is_sent=True):

try:

notification_data={

'notification_object':notification_object_id,

'notifier':notifier,

'is_sent': is_sent

}

notification_serializer = NotificationsSerializer(data=notification_data)

if notification_serializer.is_valid():

notification_serializer.save()

return True, {"notification created"}

else:

responseData = {"message": "Invalid request.", "status":False, "status_code":status.HTTP_300_MULTIPLE_CHOICES ,"dev_message": "invalid case id"}

return False, responseData

except Exception as e:

log.error("Exception in create_notification: " + str(e))

return False, {"message": "Exception", "dev_message": str(e), "status_code": status.HTTP_300_MULTIPLE_CHOICES, "status": False}

Next we need to call first one to create the notification for appropriate type like:

send_notification_type = "file_upload"
status_notifiaction_obj,notification_object_id = create_notification_object(send_notification_type,entity_id=file_object_id,actor=user_id)

1 2	send_notification_type = "file_upload" status_notifiaction_obj,notification_object_id = create_notification_object(send_notification_type,entity_id=file_object_id,actor=user_id)

Next we fetch all the users(or members to whom notifications be sent) and and notifications for each of them like:

if status_notification_obj:
    for member in members_list:
        notification_status,notification = create_notification(notification_object_id,notifier=member)

if status_notification_obj:

for member in members_list:

notification_status,notification = create_notification(notification_object_id,notifier=member)