Chanx - A comprehensive toolkit for Django Channels with DRF-style authentication, structured messaging, and developer tools

[huynguyengl99]

Chanx (CHANnels-eXtension) - A Production-Ready Django Channels Toolkit

Hi Django Channels community! 👋

I've been working on Chanx, a comprehensive toolkit that extends Django Channels with features that production applications commonly need. I wanted to share it with the community and get your feedback.

What is Chanx?

Chanx fills the gaps between Django Channels' core infrastructure and what you need for production WebSocket applications. While Channels provides excellent WebSocket support, it leaves implementation details like authentication, message validation, and testing utilities to developers.

Key Features

• 🔐 DRF-Style Authentication: Use the same authentication and permission classes from Django REST Framework
• 📝 Type-Safe Messaging: Pydantic-based message validation with discriminated unions and generic type parameters
• 🎮 WebSocket Playground: Interactive UI for testing WebSocket endpoints (like Swagger for WebSockets)
• 👥 Group Management: Simplified pub/sub messaging with automatic group handling
• 📡 Typed Channel Events: Type-safe channel layer events for inter-consumer communication
• 🛣️ Enhanced Routing: Django-style path(), re_path(), and include() functions designed for WebSocket routing
• 🧪 Testing Utilities: Specialized tools for testing WebSocket consumers, including multi-user scenarios
• 🔍 Full Type Safety: Complete mypy/pyright support with generic type parameters

Quick Example

from typing import Any, Literal
from chanx.generic.websocket import AsyncJsonWebsocketConsumer
from chanx.messages.base import BaseMessage
from rest_framework.authentication import SessionAuthentication
from rest_framework.permissions import IsAuthenticated

class ChatMessage(BaseMessage):
    action: Literal["chat"] = "chat"
    payload: str

class ChatConsumer(AsyncJsonWebsocketConsumer[ChatMessage]):
    authentication_classes = [SessionAuthentication]
    permission_classes = [IsAuthenticated]
    
    async def build_groups(self) -> list[str]:
        room_id = self.scope["url_route"]["kwargs"]["room_id"]
        return [f"chat_room_{room_id}"]
    
    async def receive_message(self, message: ChatMessage, **kwargs: Any) -> None:
        # Type-safe message handling with pattern matching
        match message:
            case ChatMessage(payload=text):
                # Broadcast to group with automatic metadata
                await self.send_group_message(message)

Why I Built This

After building several WebSocket applications with Channels, I found myself repeatedly implementing:

• Authentication bridges between DRF and WebSocket consumers
• Message validation and type safety
• Group messaging patterns
• Testing utilities for multi-user scenarios
• Developer tooling for debugging WebSocket connections

Chanx consolidates these patterns into a cohesive framework that works seamlessly with Django's ecosystem.

Generic Type System

One of the unique features is the generic type system that provides compile-time type safety:

# Three generic parameters for different use cases:
class MyConsumer(AsyncJsonWebsocketConsumer[
    IncomingMessage,  # Required: Union of incoming message types
    ChannelEvent,     # Optional: Union of channel event types  
    Room             # Optional: Model for object-level permissions
]):
    queryset = Room.objects.all()  # For object-level permissions
    
    async def receive_message(self, message: IncomingMessage, **kwargs: Any) -> None:
        # message is properly typed
        ...
    
    async def receive_event(self, event: ChannelEvent) -> None:
        # Handle typed channel events
        ...

WebSocket Playground

The included playground provides a Swagger-like interface for WebSocket endpoints:

• Auto-discovers available WebSocket routes
• Interactive connection testing with authentication
• Message composer with example generation
• Real-time message history with syntax highlighting

Testing Made Easy

from chanx.testing import WebsocketTestCase

class ChatTest(WebsocketTestCase):
    ws_path = "/ws/chat/room1/"
    
    async def test_multi_user_chat(self):
        # Test with multiple users
        user1_comm = self.auth_communicator
        user2_comm = self.create_communicator(headers=user2_headers)
        
        await user1_comm.connect()
        await user2_comm.connect()
        
        # Test group messaging
        await user1_comm.send_message(ChatMessage(payload="Hello!"))
        responses = await user2_comm.receive_all_json(wait_group=True)
        
        assert responses[0]["payload"] == "Hello!"
        assert responses[0]["is_mine"] == False

Links

• GitHub: https://github.com/huynguyengl99/chanx
• Documentation: https://chanx.readthedocs.io/
• PyPI: https://pypi.org/project/chanx/
• Complete Example: https://github.com/huynguyengl99/chanx-example

Installation

pip install chanx

Looking for Feedback

I'd love to hear your thoughts on:

• Does this solve problems you've encountered with Channels?
• Are there features you'd like to see added?
• How does the API design feel to you?
• Any suggestions for improving the developer experience?

The project is actively maintained and I'm always looking to improve it based on real-world usage and community feedback.

Thanks for checking it out! 🚀

[carltongibson]

Looks cool @huynguyengl99