Skip to content

Integrating WebSockets with Django Channels

Django Channels extends Django to handle WebSockets, long-poll HTTP, and other asynchronous protocols, enabling real-time features in Django apps.

This guide covers basic setup, routing, consumers, and best practices for WebSocket support using Django Channels.

🚀 1. Installation

Install Django Channels and its dependencies:

pip install channels

Add Channels to your INSTALLED_APPS in settings.py:

INSTALLED_APPS = [
    # ...
    "channels",
]

Set the ASGI application:

ASGI_APPLICATION = "your_project.asgi.application"

🏗 2. ASGI Configuration

Create or update asgi.py to include Channels routing:

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

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "your_project.settings")

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

📡 3. Routing

Create a routing.py file in your Django app:

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r"ws/chat/(?P<room_name>\w+)/$", consumers.ChatConsumer.as_asgi()),
]

🛠 4. Creating a Consumer

Consumers handle WebSocket connections similar to Django views.

Example consumers.py:

import json
from channels.generic.websocket import AsyncWebsocketConsumer

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
        self.room_group_name = f"chat_{self.room_name}"

        # Join room group
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name,
        )
        await self.accept()

    async def disconnect(self, close_code):
        # Leave room group
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name,
        )

    # Receive message from WebSocket
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json.get("message")

        # Broadcast message to group
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                "type": "chat_message",
                "message": message,
            },
        )

    # Receive message from group
    async def chat_message(self, event):
        message = event["message"]

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

🔐 5. Authentication & Permissions

Use Django’s standard authentication system. Channels provides AuthMiddlewareStack to populate scope["user"] with the authenticated user.

Check authentication inside the consumer:

async def connect(self):
    if self.scope["user"].is_anonymous:
        await self.close()
    else:
        await self.accept()

⚠️ 6. Best Practices

  • Use groups to broadcast messages efficiently.
  • Handle disconnects to clean up resources.
  • Avoid blocking code; use async features fully.
  • Use Channels layers like Redis for multi-instance support.
  • Sanitize and validate incoming messages.

🔚 Next Steps