WebSockets with Django Channels

HTTP is a conversation where the browser always speaks first: it asks, the server answers, and the connection is done. That is a problem for the library's front desk screen, which should show the number of copies on the shelf the moment a loan happens anywhere in the building. Asking the server every two seconds ("polling") wastes requests and is still late. A WebSocket is a long-lived, two-way connection: once open, either side can send a message at any time.

Plain Django handles HTTP. Django Channels, an official Django project maintained separately from Django itself, adds WebSockets (and other long-lived protocols) on top of ASGI. In this lesson you will build a live "shelf" feed: a consumer that accepts WebSocket connections, a group that all screens join, a broadcast from the loan service, an origin check, a test, and the production setup. You should have read Async Views first.

How Channels fits in

Browser --ws://.../ws/shelf/--> ASGI server (Daphne or Uvicorn)
                                   |
                         ProtocolTypeRouter
                     http |               | websocket
               normal Django views    AuthMiddlewareStack -> URLRouter -> ShelfConsumer
                                                                 |
                                          channel layer (Redis): group "shelf"
                                                                 |
                 loan service --group_send("shelf", ...)--> every connected screen

• A consumer is to a WebSocket what a view is to an HTTP request: a class with methods for "connected", "message received" and "disconnected". Unlike a view, one consumer instance lives as long as the connection.
• The channel layer is a message bus between processes. Each connection has a unique channel name, and channels can join named groups. Sending to a group reaches every member, even consumers running in other worker processes.
• The scope holds connection details, like request does for views: the path, headers and, with AuthMiddlewareStack, the logged-in user from the session cookie.

Step by step: a live shelf feed

1. Install the packages: python -m pip install channels daphne. Daphne is the ASGI server from the Channels project.
2. In settings, put "daphne" first in INSTALLED_APPS (it replaces runserver with an ASGI version), add your app, and set ASGI_APPLICATION and a channel layer:

# config/settings/base.py
INSTALLED_APPS = [
    "daphne",
    "django.contrib.admin",
    # ... the rest as before
    "live",
]
ASGI_APPLICATION = "config.asgi.application"
CHANNEL_LAYERS = {"default": {"BACKEND": "channels.layers.InMemoryChannelLayer"}}   # development only

3. Write the consumer. AsyncJsonWebsocketConsumer converts JSON text frames to Python dictionaries and back:

# live/consumers.py
from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer

from catalog.models import Book

SHELF_GROUP = "shelf"


class ShelfConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        self.user = self.scope["user"]
        await self.channel_layer.group_add(SHELF_GROUP, self.channel_name)
        await self.accept()
        await self.send_json({"type": "hello", "user": self.user.get_username() or "guest"})

    async def disconnect(self, code):
        await self.channel_layer.group_discard(SHELF_GROUP, self.channel_name)

    async def receive_json(self, content):
        if content.get("action") == "ask":
            available = await self.available(content["slug"])
            await self.send_json({"type": "stock", "slug": content["slug"], "available": available})

    async def stock_changed(self, event):            # handles {"type": "stock.changed"}
        await self.send_json({"type": "stock", "slug": event["slug"], "available": event["available"]})

    @database_sync_to_async
    def available(self, slug):
        return Book.objects.values_list("available", flat=True).get(slug=slug)

4. Route WebSocket URLs and build the ASGI application. Note the order: Django is set up with get_asgi_application() before anything imports models:

# live/routing.py
from django.urls import path

from . import consumers

websocket_urlpatterns = [
    path("ws/shelf/", consumers.ShelfConsumer.as_asgi()),
]


# config/asgi.py
import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.dev")
django_asgi_app = get_asgi_application()        # set up Django before importing consumers

from channels.auth import AuthMiddlewareStack                       # noqa: E402
from channels.routing import ProtocolTypeRouter, URLRouter          # noqa: E402
from channels.security.websocket import AllowedHostsOriginValidator  # noqa: E402

from live.routing import websocket_urlpatterns                      # noqa: E402

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": AllowedHostsOriginValidator(
        AuthMiddlewareStack(URLRouter(websocket_urlpatterns))
    ),
})

5. Broadcast from ordinary, synchronous code such as the loan service. group_send() is async, so wrap it with async_to_sync. The "type" key names the consumer method to call: dots become underscores, so "stock.changed" runs stock_changed():

# live/broadcast.py
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer

from .consumers import SHELF_GROUP


def announce_stock(slug, available):
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        SHELF_GROUP, {"type": "stock.changed", "slug": slug, "available": available})


# loans/services.py, at the end of borrow_book()
transaction.on_commit(lambda: announce_stock(book.slug, book.available))

6. Run the development server as usual. With daphne installed it now speaks ASGI:

python manage.py runserver 8765

System check identified no issues (0 silenced).
November 02, 2026 - 13:57:54
Django version 6.1.2, using settings 'config.settings.dev'
Starting ASGI/Daphne version 4.2.3 development server at http://127.0.0.1:8765/
Quit the server with CTRL-BREAK.

In the browser, a few lines of JavaScript connect and react to messages: const ws = new WebSocket("ws://" + location.host + "/ws/shelf/"); then ws.onmessage = (e) => { const msg = JSON.parse(e.data); ... }. Use wss:// on HTTPS sites.

Test it without a browser

Channels includes WebsocketCommunicator, which talks to your ASGI application directly, like Django's test client does for HTTP. Here two screens (the desk and a phone) connect, the phone asks about one book, the loan service broadcasts a change, and a connection from a foreign website tries to sneak in:

import asyncio

from asgiref.sync import sync_to_async
from channels.testing import WebsocketCommunicator

from config.asgi import application
from live.broadcast import announce_stock

ORIGIN = [(b"origin", b"http://localhost:8000")]


async def main():
    desk = WebsocketCommunicator(application, "/ws/shelf/", headers=ORIGIN)
    phone = WebsocketCommunicator(application, "/ws/shelf/", headers=ORIGIN)
    print("desk connected:", (await desk.connect())[0])
    print("phone connected:", (await phone.connect())[0])
    print("desk got:", await desk.receive_json_from())
    print("phone got:", await phone.receive_json_from())

    await phone.send_json_to({"action": "ask", "slug": "godaan"})
    print("phone got:", await phone.receive_json_from())

    await sync_to_async(announce_stock)("malgudi-days", 5)     # what the loan service does
    print("desk got:", await desk.receive_json_from())
    print("phone got:", await phone.receive_json_from())
    await desk.disconnect()
    await phone.disconnect()

    evil = WebsocketCommunicator(application, "/ws/shelf/", headers=[(b"origin", b"https://evil.example")])
    print("evil origin connected:", (await evil.connect())[0])

asyncio.run(main())

desk connected: True
phone connected: True
desk got: {'type': 'hello', 'user': 'guest'}
phone got: {'type': 'hello', 'user': 'guest'}
phone got: {'type': 'stock', 'slug': 'godaan', 'available': 1}
desk got: {'type': 'stock', 'slug': 'malgudi-days', 'available': 5}
phone got: {'type': 'stock', 'slug': 'malgudi-days', 'available': 5}
evil origin connected: False

One group_send() reached both screens. The last line shows AllowedHostsOriginValidator at work: browsers send an Origin header with WebSocket connections, and the validator refuses origins that are not in ALLOWED_HOSTS. WebSockets are not covered by Django's CSRF protection, so without this check any website could open a connection using your visitor's session cookie. In pytest, the Channels documentation suggests the pytest-asyncio plugin: the same code goes in an async def test marked with @pytest.mark.asyncio, plus pytest-django's django_db marker when the consumer reads the database.

Production setup

• Channel layer: InMemoryChannelLayer works only inside one process, so a broadcast from one worker never reaches screens connected to another. In production install channels-redis and use "BACKEND": "channels_redis.core.RedisChannelLayer" with "CONFIG": {"hosts": [os.environ["REDIS_URL"]]}.
• Server: run the ASGI app with daphne -b 127.0.0.1 -p 8001 config.asgi:application, or with Uvicorn as in Deployment. Some teams keep Gunicorn for normal pages and send only /ws/ to the ASGI server.
• Nginx: a WebSocket starts as an HTTP request that asks to "upgrade". Nginx must pass that through for the /ws/ location: proxy_http_version 1.1;, proxy_set_header Upgrade $http_upgrade; and proxy_set_header Connection "upgrade";. Raise proxy_read_timeout, because quiet WebSockets otherwise get cut after a minute.
• Reconnects: connections drop on mobile networks all the time. The browser code should reconnect with a growing delay and ask for current values after reconnecting, because messages sent while it was away are not stored.

Common mistakes

• Forgetting "daphne" at the top of INSTALLED_APPS (or not setting ASGI_APPLICATION). runserver stays WSGI and WebSocket connections fail.
• Importing models in asgi.py before get_asgi_application(). Django raises AppRegistryNotReady.
• Calling the sync ORM inside async consumer methods. Wrap it with database_sync_to_async, or use the async ORM methods.
• Using InMemoryChannelLayer in production. Broadcasts stay inside one process; use Redis.
• No origin check. Wrap the WebSocket router in AllowedHostsOriginValidator.

Key points

• WebSockets keep a two-way connection open; Channels adds them to Django through ASGI.
• A consumer handles connect, receive and disconnect; route it with URLRouter inside ProtocolTypeRouter.
• Groups on a channel layer let any code broadcast to many connections with group_send(); the message type picks the consumer method.
• AuthMiddlewareStack puts the session user in scope["user"]; AllowedHostsOriginValidator blocks other websites.
• Test with WebsocketCommunicator; in production use a Redis channel layer, an ASGI server and Nginx upgrade headers.

class ShelfConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        self.user = self.scope["user"]
        if not self.user.is_authenticated or not self.user.is_staff:
            await self.close(code=4403)
            return
        await self.channel_layer.group_add(SHELF_GROUP, self.channel_name)
        await self.accept()
        await self.send_json({"type": "hello", "user": self.user.get_username()})


async def check():
    guest = WebsocketCommunicator(application, "/ws/shelf/", headers=ORIGIN)
    print(await guest.connect())

(False, 4403)