Django Channels Integration

Chanx provides seamless integration with Django Channels, including WebSocket consumers, AsyncAPI documentation views, DRF authentication, and testing utilities.

WebSocket Consumer

class chanx.channels.websocket.AsyncJsonWebsocketConsumer(*args: Any, **kwargs: Any)

Django Channels WebSocket consumer with Chanx enhanced features.

Combines Chanx's automatic message handling, type validation, authentication, and group broadcasting capabilities with Django Channels' WebSocket consumer.

Features from Chanx mixin:

Automatic message type discovery from @ws_handler decorators
Type-safe message validation using Pydantic discriminated unions
Built-in authentication with Django REST Framework integration
Group broadcasting and channel event system
Comprehensive error handling and logging

Features from Django Channels:

Django ASGI integration
Django channel layer support (Redis, in-memory, etc.)
Django authentication and session support
WebSocket lifecycle management

get_channel_layer(): Returns a channel layer by alias, or None if it is not configured.

Views

class chanx.channels.views.AsyncAPISchemaView(**kwargs)

Django view to generate AsyncAPI schema from WebSocket routing.

This view introspects the ASGI application routing to find all WebSocket consumers, extracts their handler information, and generates an AsyncAPI 3.0 schema. Supports both JSON and YAML output formats.

get(request: HttpRequest) → HttpResponse: Generate and return AsyncAPI schema in requested format (JSON or YAML).

class chanx.channels.views.AsyncAPIDocsView(**kwargs)

Django view to render AsyncAPI documentation with interactive UI.

This view renders the AsyncAPI schema in a user-friendly HTML interface using AsyncAPI's official HTML template or a custom template.

get(request: HttpRequest) → HttpResponse: Render AsyncAPI documentation page.

chanx.channels.views.generate_asyncapi_schema(request: HttpRequest) → dict[str, Any]

Generate AsyncAPI 3.0 schema using the new generator system.

Parameters:: request -- The HTTP request object to determine WebSocket base URL
Returns:: AsyncAPI schema dictionary

Management Commands

Generate AsyncAPI schema files without requiring an HTTP server.

Command:

python manage.py generate_asyncapi_schema [options]

Parameters:

Option	Default	Description
`--format {json,yaml}`	`yaml`	Output format for the schema file
`--file FILE`	`stdout`	Output file path. If not specified, outputs to stdout
`--base-url BASE_URL`	`settings` or `ws://localhost:8000`	WebSocket base URL (e.g., `ws://localhost:8000` or `wss://api.example.com`)
`--title TITLE`	`settings.ASYNCAPI_TITLE`	API title for AsyncAPI documentation
`--api-version VERSION`	`settings.ASYNCAPI_VERSION`	API version for AsyncAPI documentation
`--description DESCRIPTION`	`settings.ASYNCAPI_DESCRIPTION`	API description for AsyncAPI documentation
`--discovery-class CLASS`	`DjangoRouteDiscovery`	Python path to custom route discovery class

Examples:

# Generate YAML to file
python manage.py generate_asyncapi_schema --format yaml --file schema.yaml

# Generate JSON to stdout
python manage.py generate_asyncapi_schema --format json

# With custom metadata
python manage.py generate_asyncapi_schema \
    --title "Production API" \
    --api-version "2.0.0" \
    --base-url wss://api.example.com \
    --file api.yaml

Authenticator

Django REST Framework integration for WebSocket authentication.

class chanx.channels.authenticator.DjangoAuthenticator(send_message: Callable[[BaseMessage], Awaitable[None]])

Authenticator for Chanx WebSocket connections.

Uses Django REST Framework authentication classes and permissions to authenticate WebSocket connections with consistent behavior to RESTful APIs.

authentication_classes

DRF authentication classes for connection verification

Type:: collections.abc.Sequence[type[rest_framework.authentication.BaseAuthentication]]

permission_classes

DRF permission classes for connection authorization

Type:: Sequence[_PermissionClass]

queryset

QuerySet or Manager used for retrieving objects, or True if no objects needed

Type:: django.db.models.query.QuerySet | django.db.models.manager.Manager | None

auth_method

HTTP verb to emulate for authentication

Type:: Literal['get', 'post', 'put', 'patch', 'delete', 'options']

auth_view_class

The view class to use for authentication (default: ChanxAuthView)

Type:: type[rest_framework.generics.GenericAPIView]

override_http_methods

Whether to override HTTP methods (get, post, etc.) of custom auth_view_class to prevent side effects during authentication (default: True). Set to False if you want to call the real action methods, but this is usually not desired as it may execute unintended operations.

Type:: bool

lookup_field

Field name for object lookup (default: 'pk')

Type:: str

lookup_url_kwarg

URL kwarg name for lookup (default: None, uses lookup_field)

Type:: str | None

Configuration:: Set any of these attributes (authentication_classes, permission_classes, queryset, lookup_field, lookup_url_kwarg) on the authenticator class or instance to override the auth_view_class defaults. You can also override get_queryset() and get_object() methods for custom logic.

auth_view_class: alias of ChanxAuthView

class chanx.channels.authenticator.ChanxAuthView(**kwargs)

Base authentication view for Chanx websockets.

Provides a REST-like interface for WebSocket authentication with Django REST Framework authentication and permissions.

serializer_class: alias of ChanxSerializer

Testing

class chanx.channels.testing.WebsocketCommunicator(application: Any, path: str, headers: list[tuple[bytes, bytes]] | None = None, subprotocols: list[str] | None = None, spec_version: int | None = None, *, consumer: type[ChanxWebsocketConsumerMixin[Any]])

Base WebSocket communicator for testing Chanx consumers with Django Channels.

Combines Chanx testing features (send_message, receive_all_messages, message validation) with Django Channels' WebSocket communicator (connect, disconnect, send_json_to).

For Django-specific features like authentication, use DjangoWebsocketCommunicator.

class chanx.channels.testing.DjangoWebsocketCommunicator(application: Any, path: str, headers: list[tuple[bytes, bytes]] | None = None, subprotocols: list[str] | None = None, spec_version: int | None = None, *, consumer: type[ChanxWebsocketConsumerMixin[Any]])

Extends WebsocketCommunicator with Django authentication and settings integration.

Adds Django-specific features:

wait_for_auth(): Handle Django authentication messages
assert_authenticated_status_ok(): Validate DRF authentication status
SEND_AUTHENTICATION_MESSAGE and CAMELIZE settings support

Use this for Django consumers with authentication.

async assert_authenticated_status_ok(max_auth_time: float = 0.5) → None

Assert that the WebSocket connection was authenticated successfully.

Waits for an authentication message and verifies that its status code is 200 OK.

Parameters:: max_auth_time -- Maximum time to wait for authentication message (in seconds)
Raises:: AssertionError -- If the authentication status is not 200 OK

async wait_for_auth(send_authentication_message: bool | None = None, max_auth_time: float = 0.5, after_auth_time: float = 0.1) → AuthenticationMessage | None

Waits for and returns an authentication message if enabled in settings.

Parameters:

send_authentication_message -- Whether to expect auth message, defaults to setting
max_auth_time -- Maximum time to wait for authentication (in seconds)
after_auth_time -- Wait time sleep after authentication (in seconds)

Returns:

Authentication message or None if auth is disabled

class chanx.channels.testing.WebsocketTestCase(*args: Any, **kwargs: Any)

Django test case for WebSocket testing with Chanx.

Integrates Chanx WebSocket testing with Django's test framework, providing:

Django TransactionTestCase inheritance for database transaction handling
Automatic Django ASGI application discovery from routing configuration
Django-style setUp/tearDown with automatic communicator cleanup
Integration with Django authentication headers via get_ws_headers()
Support for Django-specific WebSocket subprotocols

Usage: 1. Subclass WebsocketTestCase 2. Set ws_path to your Django WebSocket endpoint 3. Override get_ws_headers() for Django authentication 4. Use self.auth_communicator for primary connection testing 5. Use create_communicator() for multi-user Django scenarios

The test case automatically discovers WebSocket routing from Django's ASGI configuration and ensures proper cleanup of all connections after each test.

property auth_communicator: DjangoWebsocketCommunicator: Returns a connected DjangoWebsocketCommunicator instance. The instance is created using create_communicator if not already exists.

create_communicator(*, router: Any | None = None, ws_path: str | None = None, headers: list[tuple[bytes, bytes]] | None = None, subprotocols: list[str] | None = None) → DjangoWebsocketCommunicator

Creates a DjangoWebsocketCommunicator for testing WebSocket connections.

Creates and tracks a communicator instance for interacting with WebSocket consumers in tests, allowing you to create multiple communicators to test various scenarios including: - Multi-user WebSocket interactions - Testing group message broadcasting - Testing authentication with different credentials - Simulating concurrent connections

The method tracks all created communicators and automatically handles their cleanup during tearDown() to prevent resource leaks.

Parameters:

router -- Application to use (defaults to self.router)
ws_path -- WebSocket path to connect to (defaults to self.ws_path)
headers -- HTTP headers to include (defaults to self.ws_headers) Use different headers for testing multiple authenticated users
subprotocols -- WebSocket subprotocols to use (defaults to self.subprotocols)

Returns:

A configured DjangoWebsocketCommunicator instance ready for connecting

Raises:

AttributeError -- If ws_path is not set and not provided

get_subprotocols() → list[str]: Returns WebSocket subprotocols to use. Override this method to provide custom subprotocols.

get_ws_headers() → list[tuple[bytes, bytes]]: Returns WebSocket headers for authentication/configuration. Override this method to provide custom headers.

setUp() → None

Set up the test environment before each test method.

Initializes WebSocket headers and subprotocols by calling the corresponding getter methods, and prepares for tracking communicators.

tearDown() → None

Clean up after each test method.

Ensures all WebSocket connections created during the test are properly disconnected to prevent resource leaks and test isolation issues.

Utils

chanx.channels.utils.asgi.get_websocket_application() → Any | None

Extract the WebSocket application from the ASGI configuration.

This function retrieves the WebSocket handler from the ASGI application, including all middleware layers like authentication, origin validation, etc.

Returns:: The WebSocket application with all middleware, or None if not found.

Settings

The Django Channels integration provides configuration through Django's settings system. All Chanx settings are contained within a single CHANX dictionary in your project's settings file.

Configuration Options

Here's an overview of all available settings and their default values:

# settings.py
CHANX = {
    'MESSAGE_ACTION_KEY': 'action',
    'SEND_COMPLETION': False,
    'SEND_MESSAGE_IMMEDIATELY': True,
    'SEND_AUTHENTICATION_MESSAGE': True,
    'CAMELIZE': False,
    'LOG_WEBSOCKET_MESSAGE': True,
    'LOG_IGNORED_ACTIONS': ['ping', 'pong'],
    'WEBSOCKET_BASE_URL': None,
    # AsyncAPI documentation settings
    'ASYNCAPI_TITLE': 'AsyncAPI Documentation',
    'ASYNCAPI_DESCRIPTION': '',
    'ASYNCAPI_VERSION': '1.0.0',
    'ASYNCAPI_SERVER_URL': None,
    'ASYNCAPI_SERVER_PROTOCOL': None,
}

Settings Details

Setting	Default	Description
`MESSAGE_ACTION_KEY`	`"action"`	Key name for the action field in messages, used as the discriminator for message types
`SEND_COMPLETION`	`False`	Whether to send completion messages after processing a client message, should set it to True for testing and False for the other environments.
`SEND_MESSAGE_IMMEDIATELY`	`True`	Whether to send message immediately rather than wait to the end of the processing to send together.
`SEND_AUTHENTICATION_MESSAGE`	`True`	Whether to send authentication status message after connection authentication
`CAMELIZE`	`False`	Whether to convert message keys to camelCase format using pyhumps. When enabled, requires either installing Chanx with the `camel-case` extra (`pip install chanx[camel-case]`) or manually installing the pyhumps package (`pip install pyhumps`).
`LOG_WEBSOCKET_MESSAGE`	`True`	Whether to log WebSocket messages (both received and sent)
`LOG_IGNORED_ACTIONS`	`[]`	List of message actions that should not be logged (e.g., frequent messages like heartbeats)
`WEBSOCKET_BASE_URL`	`None`	WebSocket URL for overriding in development or testing
`ASYNCAPI_TITLE`	`"AsyncAPI Documentation"`	Title for generated AsyncAPI documentation
`ASYNCAPI_DESCRIPTION`	`""`	Description for generated AsyncAPI documentation
`ASYNCAPI_VERSION`	`"1.0.0"`	Version for generated AsyncAPI documentation
`ASYNCAPI_SERVER_URL`	`None`	Server URL for AsyncAPI documentation (auto-detected if not set)
`ASYNCAPI_SERVER_PROTOCOL`	`None`	Server protocol for AsyncAPI documentation (auto-detected if not set)

Using Settings in Code

Chanx's settings are accessible through the chanx_settings object:

from chanx.channels.settings import chanx_settings

# Access a setting
action_key = chanx_settings.MESSAGE_ACTION_KEY

# Check if completion messages are enabled
if chanx_settings.SEND_COMPLETION:
    # Do something

Optional Dependencies

Some Chanx features require additional packages. You can install these along with Chanx using extras:

# Install with camelCase conversion support
pip install chanx[camel-case]

This installs the pyhumps package which is required when using the CAMELIZE setting. Without this package enabling the setting will raise a runtime error.