How to Add Real-Time WebSocket Features to Django with Django Channels and Redis

Mar 14, 2026

Problem

I was building a Django application that needed real-time features—a live notification system for when users complete actions, and a chat feature for customer support. Django, by default, only handles HTTP requests. It has no way to maintain persistent WebSocket connections.

Traditional Django is synchronous and WSGI-based. Each HTTP request is independent: the server receives a request, processes it, sends a response, and closes the connection. WebSockets require the opposite—a persistent connection where the server can push data to the client at any time.

I considered using a separate Node.js WebSocket server, but that meant maintaining two codebases, duplicating authentication logic, and dealing with cross-service communication. Then I discovered Django Channels.

Environment

I implemented this with the following stack:

Python: 3.11+
Django: 5.0
Django Channels: 4.0
Redis: 7.x (for channel layer)
Daphne: 4.0 (ASGI server)

What Happened When I Tried Traditional Approaches

First, I considered polling. The client repeatedly asks the server “any new notifications?” every few seconds:

Client: "Any new notifications?"
Server: "No"
Client: "Any new notifications?"
Server: "No"
Client: "Any new notifications?"
Server: "Yes! Here's one..."

This is simple but wasteful. Most requests return nothing. When I had 1000 users polling every 5 seconds, that’s 200 requests per second hitting my server—most returning empty responses. The latency is also terrible. In the worst case, a notification appears right after a poll, but the user won’t see it until the next poll cycle.

Then I looked at third-party services like Pusher or Ably. They handle the WebSocket infrastructure for you:

import pusher

pusher_client = pusher.Pusher(
    app_id='my-app-id',
    key='my-key',
    secret='my-secret',
    cluster='us2'
)

# Push notification
pusher_client.trigger('notifications', 'new-message', {
    'message': 'Hello world'
})

This works, but it adds cost, introduces an external dependency, and still requires me to set up authentication and authorization for the WebSocket connections. For a production Django app, I wanted something I could control entirely within my Django project.

How to Solve It: Django Channels + Redis

Django Channels extends Django to handle WebSockets by adding an ASGI layer alongside the traditional WSGI layer. Here’s how I set it up.

Step 1: Install Dependencies

pip install channels redis daphne

Then add channels to your INSTALLED_APPS:

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'channels',  # Add this
    'myapp',
]

Step 2: Configure ASGI Application

Django Channels requires an ASGI application instead of WSGI. I updated my settings:

# Point to the ASGI application
ASGI_APPLICATION = 'myproject.asgi.application'

# Configure channel layer with Redis
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)],
        },
    },
}

Then created the ASGI application file:

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import myapp.routing

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(
            myapp.routing.websocket_urlpatterns
        )
    ),
})

The ProtocolTypeRouter is the key—it routes HTTP requests to Django’s standard ASGI handler and WebSocket connections to our custom routing.

Step 3: Create WebSocket Consumers

Consumers are like views, but for WebSockets. I created a consumer for real-time notifications:

import json
from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import database_sync_to_async

class NotificationConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        if self.scope["user"].is_anonymous:
            # Reject anonymous users
            await self.close()
        else:
            # Accept connection and add to user-specific group
            self.user_group_name = f'user_{self.scope["user"].id}'
            await self.channel_layer.group_add(
                self.user_group_name,
                self.channel_name
            )
            await self.accept()

    async def disconnect(self, close_code):
        # Remove from group on disconnect
        await self.channel_layer.group_discard(
            self.user_group_name,
            self.channel_name
        )

    async def receive(self, text_data):
        # Handle incoming messages from WebSocket
        text_data_json = json.loads(text_data)
        message = text_data_json['message']

        # Echo back (for demonstration)
        await self.send(text_data=json.dumps({
            'message': message
        }))

    async def notification_message(self, event):
        # Handle messages from channel layer
        message = event['message']

        # Send to WebSocket
        await self.send(text_data=json.dumps({
            'message': message
        }))

Notice the notification_message method. This handles messages sent to the group from elsewhere in the Django application.

Step 4: Define WebSocket Routing

Create a routing file similar to Django’s urls.py:

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/notifications/$', consumers.NotificationConsumer.as_asgi()),
]

Step 5: Send Messages from Django Views

Now any part of my Django application can push real-time notifications. I created a utility function:

from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync

def send_notification(user_id, message):
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        f'user_{user_id}',
        {
            'type': 'notification_message',
            'message': message
        }
    )

Using it in a view:

from django.http import JsonResponse
from .notifications import send_notification

def complete_task(request):
    # Do some work...
    task_id = request.POST.get('task_id')
    # ... process the task ...

    # Notify the user in real-time
    send_notification(
        user_id=request.user.id,
        message={
            'type': 'task_complete',
            'task_id': task_id,
            'text': f'Task {task_id} completed successfully!'
        }
    )

    return JsonResponse({'status': 'ok'})

Step 6: Frontend WebSocket Client

On the frontend, I connected to the WebSocket:

const socket = new WebSocket(
    'ws://' + window.location.host + '/ws/notifications/'
);

socket.onopen = function(e) {
    console.log('WebSocket connected');
};

socket.onmessage = function(e) {
    const data = JSON.parse(e.data);
    console.log('Received:', data.message);

    // Show notification
    showNotification(data.message.text);
};

socket.onclose = function(e) {
    console.log('WebSocket closed, reconnecting...');
    setTimeout(function() {
        location.reload();
    }, 5000);
};

socket.onerror = function(e) {
    console.error('WebSocket error:', e);
};

// Send a test message
socket.send(JSON.stringify({
    'message': 'Hello from client'
}));

Step 7: Run with Daphne

Gunicorn only handles WSGI. For ASGI, I use Daphne:

daphne -b 0.0.0.0 -p 8000 myproject.asgi:application

Or with a process manager like systemd:

[Unit]
Description=Daphne ASGI Server
After=network.target

[Service]
User=www-data
Group=www-data
WorkingDirectory=/var/www/myproject
ExecStart=/var/www/myproject/venv/bin/daphne \
    -b 0.0.0.0 -p 8000 myproject.asgi:application
Restart=always

[Install]
WantedBy=multi-user.target

The Reason: Why ASGI and Channel Layers Work

Understanding the architecture helped me debug issues later.

ASGI vs WSGI

WSGI is synchronous and handles only HTTP. ASGI is asynchronous and handles multiple protocols:

WSGI: HTTP Request → Response → Done
ASGI: HTTP Request → Response → Done
      WebSocket Connect → Messages → Disconnect
      HTTP/2 Stream → Data → End

When Django receives a WebSocket connection, it doesn’t close it after one exchange. The connection stays open, and the consumer handles multiple message exchanges over time.

Channel Layer Architecture

The channel layer is what enables multiple Django instances to communicate. Redis acts as a message broker:

User A connects to Instance 1
User B triggers action → sends message to Redis
Redis broadcasts to Instance 1
Instance 1 sends WebSocket message to User A

Without the channel layer, each Django instance would be isolated. User A might connect to instance 1, but the notification might be sent from instance 2. Redis bridges this gap.

Why Redis for Production

During development, Channels includes an in-memory channel layer:

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels.layers.InMemoryChannelLayer',
    },
}

This works for single-process development but fails in production because each process has its own memory. Redis provides a shared message bus:

Instance 1 ←→ Redis ←→ Instance 2
     ↑              ↑
     └─── Shared message bus ───┘

I made this mistake initially—deploying with in-memory channel layer and wondering why messages only arrived sometimes (when the user happened to connect to the same instance as the sender).

Authentication in WebSockets

The AuthMiddlewareStack loads the Django session and authenticates the user:

# In asgi.py
"websocket": AuthMiddlewareStack(
    URLRouter(...)
)

# AuthMiddlewareStack does:
# 1. Parse session cookie from WebSocket handshake
# 2. Load Django session from database
# 3. Attach user to self.scope["user"]
# 4. Pass to URLRouter

This means the consumer has access to self.scope["user"] just like a Django view has request.user.

Common Mistakes I Made

Mistake 1: Using in-memory channel layer in production

# WRONG: Only works for single process
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels.layers.InMemoryChannelLayer',
    },
}

The fix is always using Redis in production:

# CORRECT: Redis works across processes and servers
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [os.environ.get('REDIS_URL', 'redis://localhost:6379')],
        },
    },
}

Mistake 2: Mixing sync and async code incorrectly

# WRONG: Blocking database call in async consumer
async def receive(self, text_data):
    user = User.objects.get(id=1)  # This blocks the event loop!

The fix is using database_sync_to_async:

from channels.db import database_sync_to_async

async def receive(self, text_data):
    user = await self.get_user(1)

@database_sync_to_async
def get_user(self, user_id):
    return User.objects.get(id=user_id)

Mistake 3: Not handling WebSocket disconnects

# WRONG: No cleanup on disconnect
async def connect(self):
    await self.accept()
    # User added to group but never removed

Always implement disconnect:

async def connect(self):
    self.group_name = f'notifications_{self.scope["user"].id}'
    await self.channel_layer.group_add(
        self.group_name,
        self.channel_name
    )
    await self.accept()

async def disconnect(self, close_code):
    # Clean up group membership
    await self.channel_layer.group_discard(
        self.group_name,
        self.channel_name
    )

Mistake 4: Running with Gunicorn only

gunicorn myproject.wsgi:application
# This only handles HTTP, not WebSockets!

Use Daphne or Uvicorn for ASGI:

daphne myproject.asgi:application
# or
uvicorn myproject.asgi:application

Mistake 5: Not configuring Redis connection pooling

For high-traffic applications, the default Redis connection settings can be a bottleneck:

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [{
                "address": "redis://localhost:6379",
                "max_connections": 50,  # Increase connection pool
            }],
            "capacity": 1500,  # Channel message buffer size
            "expiry": 10,  # Message expiry in seconds
        },
    },
}

Summary

In this post, I showed how to add real-time WebSocket support to Django using Django Channels and Redis:

ASGI extends Django beyond HTTP: Django Channels adds an async layer that handles WebSockets alongside normal HTTP requests
Consumers are like views for WebSockets: They handle connect, receive, and disconnect events
Channel layers enable multi-instance communication: Redis acts as a message broker, allowing any Django instance to send messages to any connected client
Production requires Redis: The in-memory channel layer only works for single-process development

The architecture insight from the Reddit discussion about PyTogether confirms this approach: “Redis for channel layers + caching + queues for workers” is the production-ready pattern. They use Django Channels for real-time collaboration features, voice/live chats, and live selections—exactly what this stack is designed for.

If you need real-time features in Django, Django Channels is the way to go. It keeps everything within the Django ecosystem, works with Django’s authentication system, and scales horizontally with Redis.

Final Words + More Resources

My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me

Here are also the most important links from this article along with some further resources that will help you in this scope:

👨‍💻 Django Channels Documentation
👨‍💻 Reddit: PyTogether - Google Docs for Python

Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!