Notifications are a key component of any modern web application, ensuring users are informed and engaged. A well-implemented notification system can handle multiple channels like in-app alerts, emails, and SMS while dynamically tailoring content for a seamless user experience. In this guide, we’ll walk you through creating a robust, scalable notification system in Django.
System Features
Our notification system is designed to provide:
- Support for Multiple Channels: Notifications via in-app alerts, email, or SMS.
- Dynamic Content Personalization: Templates with placeholders to generate personalized messages.
- Event-Based Triggers: Trigger notifications based on specific system or user events.
- Status Tracking: Monitor the delivery status for email and SMS notifications.
- Admin and System Integration: Notifications can be triggered by administrators or system events.
Defining the Models
1. Notification Templates
Templates act as the backbone of our system, storing reusable content for notifications.
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)
Key Features:
-
template: Text with placeholders for dynamic values like{{username}}. -
channel: Specifies whether it’s an email, SMS, or in-app notification. -
trigger_event: Associates the template with a specific event.
2. General Notifications
The Notification model links templates to users and stores any dynamic payload for personalization.
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. Channel-Specific Models
To handle emails and SMS uniquely, we define specific models.
Email Notifications
This model manages email-specific data, such as dynamic message generation and delivery tracking.
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
SMS Notifications
Similar to email notifications, SMS-specific logic is implemented here.
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
Admin Integration
To make managing notifications easier, we register the models in the Django admin panel.
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']
Notification Service
We’ll implement a service layer to manage sending notifications through various channels.
Strategy Pattern
Using the Strategy Pattern, we’ll define classes for each notification channel.
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")
Notification Service
This service ties everything together, selecting the appropriate strategy based on the notification channel.
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)
Usage Example
Here’s how you can use the notification service:
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)
If you found this guide helpful and insightful, don’t forget to like and follow for more content like this. Your support motivates me to share more practical implementations and in-depth tutorials. Let’s keep building amazing applications together!
Top comments (0)