Channels & WebSockets

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

from . import consumers

websocket_urlpatterns = [
    path("ws/echo/", consumers.EchoConsumer.as_asgi()),
    path("ws/books/<int:book_id>/discussion/", consumers.DiscussionConsumer.as_asgi()),
    path("ws/notifications/", consumers.NotificationConsumer.as_asgi()),
]

http_urlpatterns = [
    path("feed/reviews/", consumers.ReviewFeedConsumer.as_asgi()),
]

# catalog/lowlevel.py
from channels.consumer import AsyncConsumer, SyncConsumer
from channels.exceptions import StopConsumer


class SyncEcho(SyncConsumer):
    """Raw ASGI events, handled in a thread-pool thread."""

    def websocket_connect(self, event):
        self.send({"type": "websocket.accept"})

    def websocket_receive(self, event):
        self.send({"type": "websocket.send", "text": event["text"]})

    def websocket_disconnect(self, event):
        raise StopConsumer()  # tell the server the application is finished


class AsyncEcho(AsyncConsumer):
    """The same thing with coroutines: every handler and self.send is async."""

    async def websocket_connect(self, event):
        await self.send({"type": "websocket.accept"})

    async def websocket_receive(self, event):
        await self.send({"type": "websocket.send", "text": event["text"]})

    async def websocket_disconnect(self, event):
        raise StopConsumer()

# catalog/consumers.py (excerpt)
class EchoConsumer(WebsocketConsumer):
    """Smallest useful consumer: accept, echo text back."""

    def connect(self):
        self.accept()

    def receive(self, text_data=None, bytes_data=None):
        if text_data is not None:
            self.send(text_data=text_data)

    def disconnect(self, code):
        pass

# catalog/consumers.py (excerpt)
class DiscussionConsumer(AsyncJsonWebsocketConsumer):
    """One group per book; everyone on the page sees every message."""

    MAX_LENGTH = 500

    async def connect(self):
        self.book_id = self.scope["url_route"]["kwargs"]["book_id"]
        self.user = self.scope["user"]
        self.group_name = f"book-{self.book_id}"

        if not self.user.is_authenticated:
            await self.close(code=4401)  # application-defined: "log in first"
            return
        if not await Book.objects.filter(pk=self.book_id).aexists():
            await self.close(code=4404)
            return

        await self.channel_layer.group_add(self.group_name, self.channel_name)
        await self.accept()

    async def disconnect(self, code):
        # connect() may have closed before joining; group_discard is harmless then
        if hasattr(self, "group_name"):
            await self.channel_layer.group_discard(self.group_name, self.channel_name)

    async def receive_json(self, content, **kwargs):
        text = str(content.get("text", "")).strip()
        if not text or len(text) > self.MAX_LENGTH:
            await self.send_json({"error": "text must be 1-500 characters"})
            return
        message = await DiscussionMessage.objects.acreate(
            book_id=self.book_id, author=self.user, text=text
        )
        await self.channel_layer.group_send(
            self.group_name,
            {
                "type": "discussion.message",  # -> discussion_message()
                "id": message.pk,
                "author": self.user.get_username(),
                "text": text,
            },
        )

    async def discussion_message(self, event):
        await self.send_json(
            {"id": event["id"], "author": event["author"], "text": event["text"]}
        )

// catalog/static/catalog/discussion.js
const scheme = window.location.protocol === "https:" ? "wss" : "ws";
const socket = new WebSocket(
  `${scheme}://${window.location.host}/ws/books/1/discussion/`,
);

socket.addEventListener("message", (event) => {
  const message = JSON.parse(event.data);
  const item = document.createElement("li");
  item.textContent = `${message.author}: ${message.text}`; // never innerHTML
  document.querySelector("#messages").append(item);
});

socket.addEventListener("close", (event) => {
  console.log("socket closed", event.code);
});

document.querySelector("#send").addEventListener("click", () => {
  const input = document.querySelector("#text");
  socket.send(JSON.stringify({ text: input.value }));
  input.value = "";
});

# catalog/consumers.py (excerpt)
class NotificationConsumer(JsonWebsocketConsumer):
    """Synchronous variant: one group per user, server-to-client pushes only."""

    def connect(self):
        user = self.scope["user"]
        if not user.is_authenticated:
            self.close(code=4401)
            return
        self.group_name = f"user-{user.pk}"
        # the channel layer is async-only: bridge with async_to_sync
        async_to_sync(self.channel_layer.group_add)(self.group_name, self.channel_name)
        self.accept()

    def disconnect(self, code):
        if hasattr(self, "group_name"):
            async_to_sync(self.channel_layer.group_discard)(
                self.group_name, self.channel_name
            )

    def notify(self, event):
        self.send_json({"kind": event["kind"], "detail": event["detail"]})

# catalog/consumers.py (excerpt)
class ReviewFeedConsumer(AsyncHttpConsumer):
    """Server-sent events: an HTTP response that stays open and is fed by group events.

    AsyncHttpConsumer.http_request() raises StopConsumer as soon as handle()
    returns, which would end the consumer before any group event arrives. An
    event-driven stream therefore overrides http_request() and returns from it
    without stopping; the consumer then lives until http_disconnect().
    """

    GROUP = "reviews"

    async def http_request(self, message):
        if message.get("more_body"):
            return  # a GET has no body; ignore any further chunks
        await self.channel_layer.group_add(self.GROUP, self.channel_name)
        await self.send_headers(
            headers=[
                (b"Cache-Control", b"no-cache"),
                (b"Content-Type", b"text/event-stream"),
            ]
        )
        # Headers only go out with the first body event; more_body keeps it open.
        await self.send_body(b": connected\n\n", more_body=True)

    async def review_posted(self, event):
        payload = json.dumps({"book": event["book"], "stars": event["stars"]})
        await self.send_body(f"data: {payload}\n\n".encode(), more_body=True)

    async def http_disconnect(self, message):
        await self.channel_layer.group_discard(self.GROUP, self.channel_name)
        await super().http_disconnect(message)

curl -N http://localhost:8760/feed/reviews/

: connected

data: {"book": "Dune", "stars": 5}

# config/settings.py (excerpt)
import os

REDIS_URL = os.environ.get("REDIS_URL")  # e.g. redis://redis.internal:6379/0

if REDIS_URL:
    CHANNEL_LAYERS = {
        "default": {
            "BACKEND": "channels_redis.core.RedisChannelLayer",
            "CONFIG": {
                "hosts": [REDIS_URL],
                "prefix": "bookshelf",
                "capacity": 200,
            },
        },
    }
else:
    CHANNEL_LAYERS = {
        "default": {"BACKEND": "channels.layers.InMemoryChannelLayer"},
    }

# catalog/tests/test_layers.py
from channels.exceptions import ChannelFull
from channels.layers import get_channel_layer

from .base import LayerTestCase


class ChannelLayerTests(LayerTestCase):
    async def test_point_to_point_and_groups(self):
        layer = get_channel_layer()
        a, b = await layer.new_channel(), await layer.new_channel()

        await layer.send(a, {"type": "ping", "n": 1})
        self.assertEqual(await layer.receive(a), {"type": "ping", "n": 1})

        await layer.group_add("room", a)
        await layer.group_add("room", b)
        await layer.group_send("room", {"type": "hello"})
        self.assertEqual((await layer.receive(a))["type"], "hello")
        self.assertEqual((await layer.receive(b))["type"], "hello")

        await layer.group_discard("room", b)
        await layer.group_send("room", {"type": "again"})
        self.assertEqual((await layer.receive(a))["type"], "again")

    async def test_group_names_are_validated(self):
        layer = get_channel_layer()
        with self.assertRaises(TypeError):
            await layer.group_add("not a valid name!", "specific.abc")
        with self.assertRaises(TypeError):
            await layer.group_add("x" * 100, "specific.abc")

    async def test_full_channels_refuse_messages(self):
        layer = get_channel_layer()
        channel = await layer.new_channel()
        for n in range(layer.capacity):  # the default capacity is 100
            await layer.send(channel, {"type": "fill", "n": n})
        with self.assertRaises(ChannelFull):
            await layer.send(channel, {"type": "one.too.many"})

# catalog/views.py
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.contrib.auth.decorators import login_required
from django.http import JsonResponse
from django.views.decorators.http import require_POST

from .middleware import issue_ticket


@login_required
@require_POST
def websocket_ticket(request):
    return JsonResponse({"ticket": issue_ticket(request.user)})


@login_required
@require_POST
def post_review(request):
    """A classic synchronous view that pushes an event to every open feed."""
    layer = get_channel_layer()
    async_to_sync(layer.group_send)(
        "reviews",
        {
            "type": "review.posted",
            "book": request.POST["book"],
            "stars": int(request.POST["stars"]),
        },
    )
    return JsonResponse({"queued": True}, status=202)


@login_required
@require_POST
async def announce(request):
    """The same push from an async view: just await the layer."""
    layer = get_channel_layer()
    await layer.group_send(
        f"book-{request.POST['book_id']}",
        {
            "type": "discussion.message",
            "id": 0,
            "author": "librarian",
            "text": request.POST["text"],
        },
    )
    return JsonResponse({"queued": True}, status=202)

# config/urls.py
from django.contrib import admin
from django.urls import path

from catalog import views

urlpatterns = [
    path("admin/", admin.site.urls),
    path("ws-ticket/", views.websocket_ticket, name="ws-ticket"),
    path("reviews/", views.post_review, name="post-review"),
    path("announce/", views.announce, name="announce"),
]

# catalog/workers.py
from channels.consumer import SyncConsumer

PROCESSED = []  # stand-in for real work such as resizing a cover image


class CoverConsumer(SyncConsumer):
    """Receives events of type "cover.resize" sent to the "covers" channel."""

    def cover_resize(self, event):
        PROCESSED.append(event["book_id"])

# config/asgi.py (excerpt): next to the "http" and "websocket" entries
from channels.routing import ChannelNameRouter

from catalog.workers import CoverConsumer

application = ProtocolTypeRouter(
    {
        # ... "http" and "websocket" as before ...
        "channel": ChannelNameRouter({"covers": CoverConsumer.as_asgi()}),
    }
)

# anywhere that can reach the layer
await get_channel_layer().send("covers", {"type": "cover.resize", "book_id": 7})

python manage.py runworker covers

# catalog/tests/test_stack.py (excerpt)
async def test_session_cookie_authenticates_the_socket(self):
    await self.async_client.aforce_login(self.alice)
    session_id = self.async_client.cookies[settings.SESSION_COOKIE_NAME].value
    cookie = f"{settings.SESSION_COOKIE_NAME}={session_id}".encode()
    communicator = WebsocketCommunicator(
        application, self.path(), headers=[ORIGIN, (b"cookie", cookie)]
    )
    connected, _ = await communicator.connect()
    self.assertTrue(connected)

    await communicator.send_json_to({"text": "from a cookie"})
    self.assertEqual((await communicator.receive_json_from())["author"], "alice")
    await communicator.disconnect()

# catalog/auth_checks.py
from importlib import import_module

from channels.auth import get_user
from channels.generic.websocket import AsyncJsonWebsocketConsumer
from django.conf import settings

SessionStore = import_module(settings.SESSION_ENGINE).SessionStore


async def fresh_user(scope):
    """Re-read the session from the store: was the user logged out elsewhere?

    scope["session"] was loaded once, when the socket opened; get_user(scope)
    re-checks the user row and the password hash, but not a session that a
    logout in another tab has since deleted.
    """
    store = SessionStore(scope["session"].session_key)
    return await get_user({"session": store})


class WhoAmIConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        if self.scope["user"].is_authenticated:
            await self.accept()
        else:
            await self.close(code=4401)

    async def receive_json(self, content, **kwargs):
        cached = await get_user(self.scope)
        current = await fresh_user(self.scope)
        if not current.is_authenticated:
            await self.close(code=4401)  # session gone: stop serving this socket
            return
        await self.send_json(
            {"cached": cached.get_username(), "current": current.get_username()}
        )

# catalog/tests/test_auth_checks.py (excerpt)
await self.async_client.alogout()  # e.g. "log out" clicked in another tab
await communicator.send_json_to({})
self.assertEqual((await communicator.receive_output())["type"], "websocket.close")

# catalog/middleware.py
from channels.db import database_sync_to_async
from channels.middleware import BaseMiddleware
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from django.core import signing
from urllib.parse import parse_qs

TICKET_SALT = "catalog.ws-ticket"
TICKET_MAX_AGE = 30  # seconds: a ticket is only needed to open the socket


def issue_ticket(user):
    """Called from a normal HTTP view for a logged-in user."""
    return signing.dumps({"uid": user.pk}, salt=TICKET_SALT)


@database_sync_to_async
def _user_for_ticket(ticket):
    try:
        data = signing.loads(ticket, salt=TICKET_SALT, max_age=TICKET_MAX_AGE)
    except signing.BadSignature:  # includes SignatureExpired
        return AnonymousUser()
    User = get_user_model()
    return User.objects.filter(pk=data["uid"], is_active=True).first() or AnonymousUser()


class TicketAuthMiddleware(BaseMiddleware):
    """Authenticate clients that cannot send cookies (mobile apps, scripts)."""

    async def __call__(self, scope, receive, send):
        scope = dict(scope)  # never mutate the caller's scope
        user = scope.get("user")
        if user is None or not user.is_authenticated:
            query = parse_qs(scope.get("query_string", b"").decode())
            ticket = (query.get("ticket") or [""])[0]
            if ticket:
                scope["user"] = await _user_for_ticket(ticket)
        return await super().__call__(scope, receive, send)

$ python manage.py runserver 127.0.0.1:8761
System check identified no issues (0 silenced).
October 03, 2026 - 02:03:11
Django version 6.1.1, using settings 'config.settings'
Starting ASGI/Daphne version 4.2.3 development server at http://127.0.0.1:8761/
Quit the server with CONTROL-C.

daphne -b 0.0.0.0 -p 8000 --proxy-headers config.asgi:application

location @proxy_to_app {
    proxy_pass http://channels-backend;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_redirect off;
    proxy_set_header Host $host;
}

$ python -m pip install 'uvicorn[standard]'   # [standard] adds a WebSocket implementation
$ uvicorn config.asgi:application --host 127.0.0.1 --port 8762
INFO:     ASGI 'lifespan' protocol appears unsupported.
INFO:     127.0.0.1:52359 - "WebSocket /ws/echo/" [accepted]