如何在 Django 中处理 Firebase 云消息通知？

Question

我正在设法通过 Django 向我的应用程序发送 FCM 通知。这是我推送通知的功能：

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'

现在，我将在视图中发送通知：

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

现在，在序列化程序中发布一些数据并保存之后。我希望为用户发送通知。但是，我想知道我是否处理正确，包括 2500 个并行连接和每分钟 400 个连接。

Answer 1

假设您有以下 FCM class 来推送通知：

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()

如果我们使用FCMThread(*args).run()，下面的class可以运行在不同的线程中。基于我的应用程序背后的逻辑；我将推送在数据库更新和创建时发生的通知，我通过重写 save() 方法来做到这一点。例如，每次使用 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)

如何在 Django 中处理 Firebase 云消息通知？

How to handle Firebase Cloud Messaging Notifications in Django?

python

django

firebase

python-asyncio

firebase-cloud-messaging