在 Django 中构建灵活的通知系统：综合指南

IT行业相对于一般传统行业，发展更新速度更快，一旦停止了学习，很快就会被行业所淘汰。所以我们需要踏踏实实的不断学习，精进自己的技术，尤其是初学者。今天golang学习网给大家整理了《在 Django 中构建灵活的通知系统：综合指南》，聊聊，我们一起来看看吧！

通知是任何现代 web 应用程序的关键组成部分，可确保用户了解情况并参与其中。实施良好的通知系统可以处理多个渠道，例如应用内警报、电子邮件和短信，同时动态定制内容以实现无缝的用户体验。在本指南中，我们将引导您在 django 中创建一个强大的、可扩展的通知系统。

系统功能

我们的通知系统旨在提供：

1. 支持多渠道：通过应用内提醒、电子邮件或短信发送通知。
2. 动态内容个性化：带有占位符的模板，用于生成个性化消息。
3. 基于事件的触发器：根据特定系统或用户事件触发通知。
4. 状态跟踪：监控电子邮件和短信通知的传送状态。
5. 管理和系统集成：通知可以由管理员或系统事件触发。

定义模型

1.通知模板

模板是我们系统的支柱，存储可重复使用的通知内容。

from django.db import models

class channeltype(models.textchoices):
    app = 'app', 'in-app notification'
    sms = 'sms', 'sms'
    email = 'email', 'email'


class triggeredbytype(models.textchoices):
    system = 'system', 'system notification'
    admin = 'admin', 'admin notification'


class triggerevent(models.textchoices):
    enrollment = 'enrollment', 'enrollment'
    announcement = 'announcement', 'announcement'
    promotional = 'promotional', 'promotional'
    reset_password = 'reset_password', 'reset password'

class notificationtemplate(models.model):
    title = models.charfield(max_length=255)
    template = models.textfield(help_text='use placeholders like {{username}} for personalization.')
    channel = models.charfield(max_length=20, choices=channeltype.choices, default=channeltype.app)
    triggered_by = models.charfield(max_length=20, choices=triggeredbytype.choices, default=triggeredbytype.system)
    trigger_event = models.charfield(max_length=50, choices=triggerevent.choices, help_text='event that triggers this template.')
    is_active = models.booleanfield(default=true)
    created_at = models.datetimefield(auto_now_add=true)
    updated_at = models.datetimefield(auto_now=true)

主要特点：

• 模板：带有动态值占位符的文本，例如 {{username}}.
• 渠道：指定是电子邮件、短信还是应用内通知。
• trigger_event：将模板与特定事件关联。

2.一般通知

通知模型将模板链接到用户并存储任何动态负载以进行个性化。

class notification(models.model):
    user = models.foreignkey(user, on_delete=models.cascade, related_name="notifications")
    content = models.foreignkey(notificationtemplate, on_delete=models.cascade, related_name="notifications")
    payload = models.jsonfield(default=dict, help_text="data to replace template placeholders.")
    is_read = models.booleanfield(default=false)
    created_at = models.datetimefield(auto_now_add=true)

3.特定于渠道的模型

为了独特地处理电子邮件和短信，我们定义了特定的模型。

电子邮件通知

该模型管理特定于电子邮件的数据，例如动态消息生成和传递跟踪。

class statustype(models.textchoices):
    pending = 'pending', 'pending'
    success = 'success', 'success'
    failed = 'failed', 'failed'

class emailnotification(models.model):
    user = models.foreignkey(user, on_delete=models.cascade, related_name='email_notifications')
    content = models.foreignkey(notificationtemplate, on_delete=models.cascade, related_name='email_notifications')
    payload = models.jsonfield(default=dict)
    status = models.charfield(max_length=20, choices=statustype.choices, default=statustype.pending)
    status_reason = models.textfield(null=true)

    @property
    def email_content(self):
        """
        populate the template with dynamic data from the payload.
        """
        content = self.content.template
        for key, value in self.payload.items():
            content = re.sub(
                rf"{{{{\s*{key}\s*}}}}",
                str(value),
                content,
            )
        return content

短信通知

与电子邮件通知类似，这里实现了短信特定逻辑。

class smsnotification(models.model):
    user = models.foreignkey(user, on_delete=models.cascade, related_name='sms_notifications')
    content = models.foreignkey(notificationtemplate, on_delete=models.cascade, related_name='sms_notifications')
    payload = models.jsonfield(default=dict)
    status = models.charfield(max_length=20, choices=statustype.choices, default=statustype.pending)
    status_reason = models.textfield(null=true)

    @property
    def sms_content(self):
        """
        populate the template with dynamic data from the payload.
        """
        content = self.content.template
        for key, value in self.payload.items():
            content = re.sub(
                rf"{{{{\s*{key}\s*}}}}",
                str(value),
                content,
            )
        return content

管理集成

为了更轻松地管理通知，我们在 django 管理面板中注册模型。

from django.contrib import admin
from notifier.models import notificationtemplate

@admin.register(notificationtemplate)
class notificationtemplateadmin(admin.modeladmin):
    list_display = ['title', 'channel', 'triggered_by', 'trigger_event', 'is_active']
    list_filter = ['channel', 'triggered_by', 'is_active']
    search_fields = ['title', 'trigger_event']

通知服务

我们将实现一个服务层来管理通过各种渠道发送通知。

策略模式

使用策略模式，我们将为每个通知通道定义类。

from abc import abc, abstractmethod
import logging

logger = logging.getlogger(__name__)

class notificationstrategy(abc):
    @abstractmethod
    def send(self, user, content, payload):
        pass


class appnotificationstrategy(notificationstrategy):
    def send(self, user, content, payload):
        notification = notification.objects.create(user=user, content=content, payload=payload)
        logger.info(f"in-app notification sent to {user.email}")
        return notification


class emailnotificationstrategy(notificationstrategy):
    def send(self, user, content, payload):
        notification = emailnotification.objects.create(user=user, content=content, payload=payload)
        try:
            self._send_email(user.email, content.title, notification.email_content)
            notification.status = "success"
        except exception as e:
            notification.status = "failed"
            notification.status_reason = str(e)
        notification.save()
        return notification

    def _send_email(self, to_email, subject, body):
        print(f"sending email to {to_email} with subject {subject}")
        if "@" not in to_email:
            raise valueerror("invalid email address")


class smsnotificationstrategy(notificationstrategy):
    def send(self, user, content, payload):
        notification = smsnotification.objects.create(user=user, content=content, payload=payload)
        try:
            self._send_sms(user.phone_number, notification.sms_content)
            notification.status = "success"
        except exception as e:
            notification.status = "failed"
            notification.status_reason = str(e)
        notification.save()
        return notification

    def _send_sms(self, phone_number, message):
        print(f"sending sms to {phone_number}: {message}")
        if not phone_number.isdigit():
            raise valueerror("invalid phone number")

通知服务

该服务将所有内容联系在一起，根据通知渠道选择适当的策略。

class notificationservice:
    _strategies: dict[type[channeltype], type[notificationstrategy]] = {
        channeltype.app: appnotificationstrategy,
        channeltype.email: emailnotificationstrategy,
        channeltype.sms: smsnotificationstrategy,
    }

    @classmethod
    def get_strategy(cls, instance: notificationtemplate) -> notificationstrategy:
        try:
            channel = channeltype[instance.channel]
            strategy = cls._strategies[channel]
        except keyerror:
            raise exception(f"unknown notification strategy {instance.channel}")

        return strategy()

    @classmethod
    def notify(
        cls,
        user: user,
        event: triggerevent,
        payload: dict,
    ):
        """
        automatically create and send a system-triggered notification.

        args:
            user: user instance.
            event: triggerevent type.
            payload: dynamic `dict` data for the notification.

        returns:
            result of the notification strategy.
        """

        content, _ = notificationtemplate.objects.get_or_create(
            trigger_event=event,
            triggered_by=triggeredbytype.system,
            defaults={
                "title": 'default title',
                "template": "this is a system notification.",
            },
        )
        strategy = cls.get_strategy(instance=content)
        return strategy.send(user, content, payload)

使用示例

以下是如何使用通知服务：

from notifier.services import NotificationService, TriggerEvent

user = User.objects.get(email="user@example.com")
payload = {"username": "John Doe", "course": "Python Basics"}

NotificationService.notify(user, TriggerEvent.USER_ENROLLED, payload)

如果您发现本指南很有帮助且富有洞察力，请不要忘记点赞并关注以获取更多此类内容。您的支持激励我分享更多实用的实现和深入的教程。让我们继续一起构建令人惊叹的应用程序！

终于介绍完啦！小伙伴们，这篇关于《在 Django 中构建灵活的通知系统：综合指南》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布文章相关知识，快来关注吧！