[FIXED] How to handle Firebase Cloud Messaging Notifications in Django?

Issue

I am managing to send FCM notifications to my app through Django. Here is my function for pushing notifications:

import firebase_admin
from firebase_admin import credentials, messaging

cred = credentials.Certificate("service-key.json")
firebase_admin.initialize_app(cred)

def send_push(title, msg, registration_token, dataObject=None):
    try:
        message = messaging.MulticastMessage(
            notification=messaging.Notification(
                title=title,
                body=msg,
            ),
            data=dataObject,
            tokens=registration_token,
        )

        response = messaging.send_multicast(message)
    except messaging.QuotaExceededError:
        raise f'Exceeding FCM notifications quota'

Now, I'll be sending a Notification inside a view:

class AdminChangeRole(viewsets.Viewset):
    serializer_class = SomeSerializer
    permission_classes = [IsAdminOnly]
    def post(self, request, *args, **kwargs):
        # code that will change the role of a user
        send_push("Role changed", f"You role was set to {self.role}", fcm_registration_token)
        # return Response

Now, after posting some data inside a serializer and save it. I'm hoping to send a notification for a User. However, I want to know if I am handling this correctly, including the 2500 connections in parallel and 400 connections per minute.

Solution

Suppose you have the following FCM class in order to push notifications:

import threading
from typing import Optional
from firebase_admin import messaging, credentials
import firebase_admin
from coreapp.utils.logging_helper import get_log

log = get_log()

cred = credentials.Certificate(
    "myproject-firebase-adminsdk.json"
)
firebase_admin.initialize_app(cred)


class FCMThread(threading.Thread):
    """
    :param title: Title of notification
    :param msg: Message or body of notification
    :param tokens: Tokens of the users who will receive this notification
    :param data: A dictionary of data fields (optional). All keys and values in the dictionary must be strings.
    :return -> None:
    """

    def __init__(
        self: threading.Thread,
        title: str,
        msg: str,
        tokens: list,
        data: Optional[list] = None,
    ) -> None:
        self.title = title
        self.msg = msg
        self.tokens = tokens
        self.data = data
        threading.Thread.__init__(self)

    def _push_notification(self):
        """
        Push notification messages by chunks of 500.
        """
        chunks = [self.tokens[i : i + 500] for i in range(0, len(self.tokens), 500)]
        for chunk in chunks:
            messages = [
                messaging.Message(
                    notification=messaging.Notification(self.title, self.msg),
                    token=token,
                    data=self.data,
                )
                for token in chunk
            ]
            response = messaging.send_all(messages)
            log.info(f"Number of successful notifications: {response._success_count}")
            log.info(
                f"Number of failed notifications: {len(messages) - response._success_count}"
            )

    def run(self):
        self._push_notification()

The following class can run in a different thread if we use FCMThread(*args).run(). Based on the logic behind my app; I will be pushing notifications that happens on database updates and create and I did this by overriding the save() method. For example, the following code would send a notification each time a User's role is updated with the class FCMThread.

class User(AbstractBaseUser, PermissionsMixin):
    email = models.EmailField(_("email address"), unique=True)
    name = models.CharField(_("name"), max_length=30, blank=True)
    phone_number = models.CharField(_("phone number"), max_length=20, blank=True)
    date_joined = models.DateTimeField(_("date joined"), auto_now_add=True)
    is_active = models.BooleanField(_("active"), default=True)
    is_staff = models.BooleanField(
        _("staff status"),
        default=False,
        help_text=_("Designates whether the user can log into this admin site."),
    )
    personal_bio = models.CharField(_("persion bio"), max_length=500, blank=True)
    is_verified = models.BooleanField(_("is verified"), default=False)
    is_approved = models.BooleanField(_("is approved"), default=False)
    avatar = models.ImageField(upload_to=user_avatar_path, null=True, blank=True)
    national_id_front = models.ImageField(
        null=True, blank=True, upload_to="users/documents/"
    )
    national_id_back = models.ImageField(
        null=True, blank=True, upload_to="users/documents/"
    )
    roles = models.IntegerField(
        choices=Roles.choices,
        default=Roles.NOT_ASSIGNED,
    )
    community = models.ForeignKey(
        Community,
        related_name="members",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    fcm_token = models.CharField(_("FCM Token"), max_length=200, null=True, blank=True)
    objects = UserManager()

    USERNAME_FIELD = "email"
    REQUIRED_FIELDS = []

    class Meta:
        verbose_name = _("user")
        verbose_name_plural = _("users")

    def save(self, *args, **kwargs):
        # send roles notification
        previous_role = None
        if User.objects.filter(id=self.id).exists():
            previous_role = User.objects.get(id=self.id).roles
            if self.roles != previous_role:
                log.info(f"Sending notifcation to {self.name}")
                fcm_tokens = (self.fcm_token,)
                FCMThread(
                    "Role update",
                    f"Your role has been changed from {previous_role} to {self.roles}",
                    fcm_tokens,
                ).start()
         super(User, self).save(*args, **kwargs)

Answered By - Samir Ahmane