DEV Community

Cover image for Real-Time Features in Django: How We Build Live Updates with Channels and WebSockets
Lycore Development
Lycore Development

Posted on

Real-Time Features in Django: How We Build Live Updates with Channels and WebSockets

#python #django #websockets #webdev

Most Django applications are request-response. A user does something, the browser sends a request, Django processes it, returns a response, done. That model works for 90% of what web applications need to do.

The other 10% — live dashboards, notifications, collaborative editing, order tracking, chat — requires the server to push data to the client without the client asking first. Django's standard request-response cycle cannot do this. WebSockets can.

This post covers how we add real-time features to Django projects using Django Channels, what the architecture looks like, and the patterns that have worked in production.

What Django Channels is (and what it isn't)

Django Channels extends Django to handle protocols other than HTTP — including WebSockets, but also HTTP/2 server-sent events, and background tasks. It does this by adding a layer called the Channel Layer, which is a message-passing system backed by Redis.

The mental model: instead of Django handling a request and returning a response, Channels handles a connection. The connection persists. Messages flow in both directions. Django code can send messages to that connection at any time — not just in response to something the client sent.

Channels is not a magic real-time solution. It requires:

  • A channel layer (Redis)
  • An ASGI server (Daphne or Uvicorn, not Gunicorn)
  • Consumers (the Channels equivalent of views)
  • A way for your Django application logic to trigger messages

Basic setup 

pip install channels channels-redis
Enter fullscreen mode Exit fullscreen mode 
# settings.py
INSTALLED_APPS = [
    ...
    'channels',
]

ASGI_APPLICATION = 'myproject.asgi.application'

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            'hosts': [('127.0.0.1', 6379)],
        },
    },
}
Enter fullscreen mode Exit fullscreen mode 
# myproject/asgi.py
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
        )
    ),
})
Enter fullscreen mode Exit fullscreen mode 
# myapp/routing.py
from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/orders/(?P<order_id>\d+)/$', consumers.OrderStatusConsumer.as_asgi()),
    re_path(r'ws/dashboard/$', consumers.DashboardConsumer.as_asgi()),
]
Enter fullscreen mode Exit fullscreen mode

Writing consumers

A consumer is like a view, but for WebSocket connections. It has lifecycle methods (websocket_connect, websocket_disconnect) and a method for handling incoming messages.

# myapp/consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import database_sync_to_async


class OrderStatusConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.order_id = self.scope['url_route']['kwargs']['order_id']
        self.group_name = f'order_{self.order_id}'

        # Check that the user has permission to watch this order
        user = self.scope['user']
        if not await self.user_can_view_order(user, self.order_id):
            await self.close()
            return

        # Join the group for this order
        await self.channel_layer.group_add(
            self.group_name,
            self.channel_name,
        )

        await self.accept()

        # Send current state on connect
        order = await self.get_order(self.order_id)
        await self.send(text_data=json.dumps({
            'type': 'order.status',
            'status': order.status,
            'updated_at': order.updated_at.isoformat(),
        }))

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(
            self.group_name,
            self.channel_name,
        )

    async def receive(self, text_data):
        # Handle messages from the client if needed
        pass

    # Handler called when a message is sent to the group
    async def order_status_update(self, event):
        await self.send(text_data=json.dumps({
            'type': 'order.status',
            'status': event['status'],
            'updated_at': event['updated_at'],
        }))

    @database_sync_to_async
    def get_order(self, order_id):
        from .models import Order
        return Order.objects.get(id=order_id)

    @database_sync_to_async
    def user_can_view_order(self, user, order_id):
        from .models import Order
        return Order.objects.filter(
            id=order_id,
            customer__user=user,
        ).exists()
Enter fullscreen mode Exit fullscreen mode

The @database_sync_to_async decorator is essential. Django's ORM is synchronous. Consumers run in an async context. Calling ORM methods directly from an async function will raise SynchronousOnlyOperation. Wrapping them with database_sync_to_async runs them in a thread pool.

Sending messages from Django code

The consumer receives and sends messages. But who triggers the message when an order status changes? That happens in your regular Django code — views, signals, Celery tasks.

# myapp/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from .models import Order


@receiver(post_save, sender=Order)
def order_status_changed(sender, instance, **kwargs):
    if not kwargs.get('update_fields') or 'status' not in kwargs['update_fields']:
        return

    channel_layer = get_channel_layer()

    async_to_sync(channel_layer.group_send)(
        f'order_{instance.id}',
        {
            'type': 'order.status_update',  # maps to order_status_update method
            'status': instance.status,
            'updated_at': instance.updated_at.isoformat(),
        }
    )
Enter fullscreen mode Exit fullscreen mode

async_to_sync is the bridge from synchronous Django code to the async channel layer. When an order's status changes, this signal fires, grabs the channel layer, and sends a message to the group for that order. Any consumer that has joined that group (any browser watching that order) will receive the message.

Live dashboard pattern

A dashboard that shows aggregated metrics — active orders, revenue today, error counts — uses a slightly different pattern. Instead of per-resource groups, it has a single broadcast group.

class DashboardConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        user = self.scope['user']
        if not user.is_authenticated or not user.is_staff:
            await self.close()
            return

        await self.channel_layer.group_add('dashboard', self.channel_name)
        await self.accept()

        # Send current metrics on connect
        metrics = await self.get_dashboard_metrics()
        await self.send(text_data=json.dumps({
            'type': 'dashboard.metrics',
            **metrics,
        }))

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard('dashboard', self.channel_name)

    async def dashboard_update(self, event):
        await self.send(text_data=json.dumps({
            'type': 'dashboard.metrics',
            **event['metrics'],
        }))

    @database_sync_to_async
    def get_dashboard_metrics(self):
        from django.utils import timezone
        from django.db.models import Count, Sum
        from .models import Order

        today = timezone.now().date()
        return {
            'active_orders': Order.objects.filter(status='processing').count(),
            'revenue_today': Order.objects.filter(
                created_at__date=today,
                status__in=['complete', 'shipped'],
            ).aggregate(total=Sum('total'))['total'] or 0,
        }
Enter fullscreen mode Exit fullscreen mode

And a Celery task that pushes updates every 30 seconds:

# tasks.py
from celery import shared_task
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer


@shared_task
def broadcast_dashboard_metrics():
    from django.utils import timezone
    from django.db.models import Sum
    from .models import Order

    today = timezone.now().date()
    metrics = {
        'active_orders': Order.objects.filter(status='processing').count(),
        'revenue_today': Order.objects.filter(
            created_at__date=today,
        ).aggregate(total=Sum('total'))['total'] or 0,
    }

    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        'dashboard',
        {
            'type': 'dashboard.update',
            'metrics': metrics,
        }
    )
Enter fullscreen mode Exit fullscreen mode 
# celery.py beat schedule
app.conf.beat_schedule = {
    'broadcast-dashboard-metrics': {
        'task': 'myapp.tasks.broadcast_dashboard_metrics',
        'schedule': 30.0,  # every 30 seconds
    },
}
Enter fullscreen mode Exit fullscreen mode

Frontend connection 

// dashboard.js
const socket = new WebSocket(
  `${window.location.protocol === 'https:' ? 'wss' : 'ws'}://${window.location.host}/ws/dashboard/`
);

socket.onopen = () => {
  console.log('Dashboard connected');
};

socket.onmessage = (event) => {
  const data = JSON.parse(event.data);

  if (data.type === 'dashboard.metrics') {
    document.getElementById('active-orders').textContent = data.active_orders;
    document.getElementById('revenue-today').textContent = `£${data.revenue_today.toFixed(2)}`;
  }
};

socket.onclose = (event) => {
  // Reconnect after a delay
  setTimeout(() => {
    window.location.reload();
  }, 3000);
};
Enter fullscreen mode Exit fullscreen mode

Reconnection handling is important. WebSocket connections drop — network flaps, server deployments, idle timeouts. The client should reconnect automatically, and the consumer should send current state on connect so the client is not left with stale data.

Production considerations

ASGI server: Gunicorn cannot serve WebSockets. In production, use Daphne (pip install daphne) or Uvicorn (pip install uvicorn). We typically run Uvicorn with Gunicorn as a process manager:

gunicorn myproject.asgi:application -k uvicorn.workers.UvicornWorker -w 4
Enter fullscreen mode Exit fullscreen mode

Nginx configuration: WebSocket connections need specific proxy headers:

location /ws/ {
    proxy_pass http://app;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
    proxy_read_timeout 86400;  # Keep connections alive for up to 24h
}
Enter fullscreen mode Exit fullscreen mode

Redis connection pool: With many concurrent WebSocket connections, the default Redis connection settings can become a bottleneck. Configure the pool size explicitly:

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            'hosts': [('127.0.0.1', 6379)],
            'capacity': 1500,
            'expiry': 10,
        },
    },
}
Enter fullscreen mode Exit fullscreen mode

When not to use WebSockets

WebSockets are not always the right tool. Consider server-sent events (SSE) if:

  • You only need server-to-client messages (no client-to-server over the connection)
  • You want simpler infrastructure (SSE is regular HTTP)
  • You need automatic reconnection built into the browser

Consider polling if:

  • Updates are infrequent (every few minutes)
  • You have many users but not many simultaneous connections
  • Simplicity matters more than real-time latency

WebSockets are the right choice when you need low-latency bidirectional communication — chat, collaborative editing, live tracking — and when you have enough concurrent users to justify the infrastructure complexity.

Lycore builds production Django applications for businesses — real-time systems, APIs, and custom web and mobile products. Get in touch if you're building something that needs live data.

Top comments (0)