Welcome to Chanx Documentation

CHANX (CHANnels-eXtension)

The missing toolkit for Django Channels — authentication, logging, structured messaging, and more.

Installation

pip install chanx

For complete documentation, visit chanx docs.

Introduction

Django Channels provides excellent WebSocket support for Django applications, but leaves gaps in authentication, structured messaging, and developer tooling. Chanx fills these gaps with a comprehensive toolkit that makes building WebSocket applications simpler and more maintainable.

Key Features

REST Framework Integration: Use DRF authentication and permission classes with WebSockets
Structured Messaging: Type-safe message handling with Pydantic validation
WebSocket Playground: Interactive UI for testing WebSocket endpoints
Group Management: Simplified pub/sub messaging with automatic group handling
Comprehensive Logging: Structured logging for WebSocket connections and messages
Error Handling: Robust error reporting and client feedback
Testing Utilities: Specialized tools for testing WebSocket consumers
Multi-user Testing Support: Test group broadcasting and concurrent connections
Object-level Permissions: Support for DRF object-level permission checks
Full Type Hints: Complete mypy and pyright support for better IDE integration and type safety

Core Components

AsyncJsonWebsocketConsumer: Base consumer with authentication and structured messaging
ChanxWebsocketAuthenticator: Bridges WebSockets with DRF authentication
Message System: Type-safe message classes with automatic validation
WebSocketTestCase: Test utilities for WebSocket consumers
Discriminated Union Messages: Runtime validation of message types with action discriminator

Configuration

Chanx can be configured through the CHANX dictionary in your Django settings. Below is a complete list of available settings with their default values and descriptions:

# settings.py
CHANX = {
    # Message configuration
    'MESSAGE_ACTION_KEY': 'action',  # Key name for action field in messages
    'CAMELIZE': False,  # Whether to camelize/decamelize messages for JavaScript clients

    # Completion messages
    'SEND_COMPLETION': False,  # Whether to send completion message after processing messages

    # Messaging behavior
    'SEND_MESSAGE_IMMEDIATELY': True,  # Whether to yield control after sending messages
    'SEND_AUTHENTICATION_MESSAGE': True,  # Whether to send auth status after connection

    # Logging configuration
    'LOG_RECEIVED_MESSAGE': True,  # Whether to log received messages
    'LOG_SENT_MESSAGE': True,  # Whether to log sent messages
    'LOG_IGNORED_ACTIONS': [],  # Message actions that should not be logged

    # Playground configuration
    'WEBSOCKET_BASE_URL': 'ws://localhost:8000'  # Default WebSocket URL for discovery
}

Example: Building an Assistant App

Let's create a simple assistant chatbot with authentication:

First, create a new Django app for your assistant:

python manage.py startapp assistants

Define your message types in assistants/messages/assistant.py:

from typing import Literal

from chanx.messages.base import BaseIncomingMessage, BaseMessage
from chanx.messages.incoming import PingMessage
from pydantic import BaseModel


class MessagePayload(BaseModel):
    content: str


class NewMessage(BaseMessage):
    """
    New message for assistant.
    """
    action: Literal["new_message"] = "new_message"
    payload: MessagePayload


class ReplyMessage(BaseMessage):
    action: Literal["reply"] = "reply"
    payload: MessagePayload


class AssistantIncomingMessage(BaseIncomingMessage):
    message: NewMessage | PingMessage

Create your consumer in assistants/consumers.py:

from typing import Any

from rest_framework.permissions import IsAuthenticated

from chanx.generic.websocket import AsyncJsonWebsocketConsumer
from chanx.messages.base import BaseMessage
from chanx.messages.incoming import PingMessage
from chanx.messages.outgoing import PongMessage

from assistants.messages.assistant import (
    AssistantIncomingMessage,
    MessagePayload,
    NewMessage,
    ReplyMessage,
)


class AssistantConsumer(AsyncJsonWebsocketConsumer):
    """Websocket to chat with server, like chat with chatbot system"""

    INCOMING_MESSAGE_SCHEMA = AssistantIncomingMessage
    permission_classes = [IsAuthenticated]

    async def receive_message(self, message: BaseMessage, **kwargs: Any) -> None:
        match message:
            case PingMessage():
                # Reply with a PONG message
                await self.send_message(PongMessage())
            case NewMessage(payload=new_message_payload):
                # Echo back with a reply message
                await self.send_message(
                    ReplyMessage(
                        payload=MessagePayload(
                            content=f"Reply: {new_message_payload.content}"
                        )
                    )
                )
            case _:
                pass

Set up WebSocket routing in assistants/routing.py:

from channels.routing import URLRouter

from chanx.urls import path

from assistants.consumers import AssistantConsumer

router = URLRouter(
    [
        path("", AssistantConsumer.as_asgi()),
    ]
)

Create a project-level routing file in your project's root directory (same level as urls.py) as routing.py:

from channels.routing import URLRouter

from chanx.routing import include
from chanx.urls import path, re_path

ws_router = URLRouter(
    [
        path("assistants/", include("assistants.routing")),
        # Add other WebSocket routes here
    ]
)

router = URLRouter(
    [
        path("ws/", include(ws_router)),
    ]
)

Configure your project's asgi.py to use the WebSocket routing:

import os

from channels.routing import ProtocolTypeRouter
from channels.security.websocket import OriginValidator
from channels.sessions import CookieMiddleware
from django.conf import settings
from django.core.asgi import get_asgi_application

# Set Django settings module
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "yourproject.settings")
django_asgi_app = get_asgi_application()

# Import your WebSocket routing
from yourproject.routing import router

# Set up protocol routing
routing = {
    "http": django_asgi_app,
    "websocket": OriginValidator(
        CookieMiddleware(router),
        settings.CORS_ALLOWED_ORIGINS + settings.CSRF_TRUSTED_ORIGINS,
    ),
}

application = ProtocolTypeRouter(routing)

Ensure your settings.py has the required settings:

INSTALLED_APPS = [
    # ...
    'channels',
    'chanx',
    'assistants',
    # ...
]

# For WebSocket origin validation
CSRF_TRUSTED_ORIGINS = [
    "http://localhost:8000",
    # Add other trusted origins
]

Connect from your JavaScript client:

const socket = new WebSocket('ws://localhost:8000/ws/assistants/');

// Add authentication headers
socket.onopen = function() {
    console.log('Connected to assistant');

    // Send a message
    socket.send(JSON.stringify({
        action: 'new_message',
        payload: {
            content: 'Hello assistant!'
        }
    }));
};

socket.onmessage = function(e) {
    const data = JSON.parse(e.data);

    if (data.action === 'reply') {
        console.log('Assistant replied:', data.payload.content);
    }
};

If you don't have a client application ready, you can use the WebSocket Playground (covered in the next section) to test your assistant endpoint without writing any JavaScript.

WebSocket Playground

Add the playground to your URLs:

urlpatterns = [
    path('playground/', include('chanx.playground.urls')),
]

Then visit /playground/websocket/ to explore and test your WebSocket endpoints. The playground will automatically discover all registered WebSocket routes from your routing.py file, including any nested routes from included routers.

Testing

Write tests for your WebSocket consumers:

from chanx.testing import WebsocketTestCase
from chanx.messages.incoming import PingMessage
from chanx.messages.outgoing import PongMessage

class TestChatConsumer(WebsocketTestCase):
    ws_path = "/ws/chat/room1/"

    async def test_connection_and_ping(self) -> None:
        # Connect and authenticate
        await self.auth_communicator.connect()
        await self.auth_communicator.assert_authenticated_status_ok()

        # Test ping/pong functionality
        await self.auth_communicator.send_message(PingMessage())
        messages = await self.auth_communicator.receive_all_json()
        assert messages == [PongMessage().model_dump()]

    async def test_multi_user_scenario(self) -> None:
        # Create communicators for multiple users
        first_comm = self.auth_communicator
        second_comm = self.create_communicator(headers=self.get_headers_for_user(user2))

        # Connect both
        await first_comm.connect()
        await second_comm.connect()

        # Test group broadcasting
        # ...

Contents

Getting Started

User Guide

Examples

Development