Implementing Real-Time Notifications with Django Channels and WebSockets

In today’s fast-paced digital world, users expect real-time updates and notifications. Whether it’s a new message, a like on a post, or a live chat, real-time functionality is essential for modern web applications. Django, a powerful Python web framework, can handle real-time features using Django Channels and WebSockets.

In this blog post, we’ll walk through how to implement real-time notifications in a Django application. By the end, you’ll have a working example of real-time notifications using Django Channels.

Prerequisites

Before we begin, ensure you have the following:

• Python 3.7+ installed
• Django installed (pip install django)
• Django Channels installed (pip install channels)
• Basic knowledge of Django (models, views, templates)

Step 1: Set Up a Django Project

1. Create a new Django project:

   django-admin startproject realtime_notifications
   cd realtime_notifications
   

2. Create a new app:

   python manage.py startapp notifications
   

3. Add notifications and channels to INSTALLED_APPS in settings.py:

   INSTALLED_APPS = [
       ...
       'notifications',
       'channels',
   ]
   

Step 2: Configure Django Channels

1. Update settings.py to use Channels as the default ASGI application:

   ASGI_APPLICATION = 'realtime_notifications.asgi.application'
   

2. Install Redis as the channel layer backend (for production):

   pip install channels_redis
   

3. Configure the channel layer in settings.py:

   CHANNEL_LAYERS = {
       'default': {
           'BACKEND': 'channels_redis.core.RedisChannelLayer',
           'CONFIG': {
               'hosts': [('127.0.0.1', 6379)],
           },
       },
   }
   

Step 3: Create a Notification Model

In notifications/models.py, define a simple Notification model:

from django.db import models
from django.contrib.auth.models import User

class Notification(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    message = models.CharField(max_length=255)
    read = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.message

Run migrations to create the database table:

python manage.py makemigrations
python manage.py migrate

Step 4: Set Up WebSocket Consumers

1. Create a consumers.py file in the notifications app:

   import json
   from channels.generic.websocket import AsyncWebsocketConsumer
   
   class NotificationConsumer(AsyncWebsocketConsumer):
       async def connect(self):
           self.user = self.scope['user']
           self.group_name = f'notifications_{self.user.id}'
   
           # Join the group
           await self.channel_layer.group_add(
               self.group_name,
               self.channel_name
           )
           await self.accept()
   
       async def disconnect(self, close_code):
           # Leave the group
           await self.channel_layer.group_discard(
               self.group_name,
               self.channel_name
           )
   
       async def send_notification(self, event):
           # Send notification to WebSocket
           await self.send(text_data=json.dumps(event['message']))
   

2. Update routing.py to route WebSocket connections:

   from django.urls import re_path
   from . import consumers
   
   websocket_urlpatterns = [
       re_path(r'ws/notifications/(?P<user_id>\w+)/$', consumers.NotificationConsumer.as_asgi()),
   ]
   

3. Update the project’s asgi.py to include WebSocket routing:

   import os
   from django.core.asgi import get_asgi_application
   from channels.routing import ProtocolTypeRouter, URLRouter
   from notifications import routing
   
   os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'realtime_notifications.settings')
   
   application = ProtocolTypeRouter({
       'http': get_asgi_application(),
       'websocket': URLRouter(routing.websocket_urlpatterns),
   })
   

Step 5: Create Views and Templates

1. In notifications/views.py, create a view to send notifications:

   from django.shortcuts import render
   from django.contrib.auth.decorators import login_required
   from channels.layers import get_channel_layer
   from asgiref.sync import async_to_sync
   from .models import Notification
   
   @login_required
   def send_notification(request):
       if request.method == 'POST':
           message = request.POST.get('message')
           notification = Notification.objects.create(user=request.user, message=message)
   
           # Send notification via WebSocket
           channel_layer = get_channel_layer()
           async_to_sync(channel_layer.group_send)(
               f'notifications_{request.user.id}',
               {
                   'type': 'send_notification',
                   'message': {'message': message}
               }
           )
           return render(request, 'notifications/send.html', {'success': True})
       return render(request, 'notifications/send.html')
   

2. Create a template (notifications/templates/notifications/send.html):

   <!DOCTYPE html>
   <html>
   <head>
       <title>Send Notification</title>
   </head>
   <body>
       <h1>Send Notification</h1>
       {% if success %}
           <p>Notification sent!</p>
       {% endif %}
       <form method="post">
           {% csrf_token %}
           <textarea name="message" placeholder="Enter your message"></textarea>
           <button type="submit">Send</button>
       </form>
   </body>
   </html>
   

Step 6: Test the Application

1. Run the development server:

   python manage.py runserver
   

2. Open the browser and navigate to the notification page.

3. Send a notification and see it appear in real-time!

This is just the beginning—you can extend this functionality to include live chat, real-time updates, and more. Django Channels opens up a world of possibilities for building dynamic, real-time web applications.