Generic Module

WebSocket Consumers

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

Base class for asynchronous JSON WebSocket consumers with authentication and permissions.

Provides DRF-style authentication/permissions, structured message handling with Pydantic validation, logging, and error handling. Subclasses must implement receive_message and set INCOMING_MESSAGE_SCHEMA.

For group messaging functionality, subclasses should also define OUTGOING_GROUP_MESSAGE_SCHEMA to enable proper validation and handling of group message broadcasts.

authentication_classes

DRF authentication classes for connection verification

Type:

collections.abc.Sequence[type[rest_framework.authentication.BaseAuthentication]] | None

permission_classes

DRF permission classes for connection authorization

Type:

collections.abc.Sequence[type[rest_framework.permissions.BasePermission] | rest_framework.permissions.OperandHolder | rest_framework.permissions.SingleOperandHolder] | None

queryset

QuerySet or Manager used for retrieving objects

Type:

Literal[True] | django.db.models.query.QuerySet | django.db.models.manager.Manager

auth_method

HTTP verb to emulate for authentication

Type:

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

authenticator_class

Class to use for performing websocket authentication, defaults to ChanxWebsocketAuthenticator

Type:

type[Any]

send_completion

Whether to send completion message after processing

Type:

bool | None

send_message_immediately

Whether to yield control after sending messages

Type:

bool | None

log_received_message

Whether to log received messages

Type:

bool | None

log_sent_message

Whether to log sent messages

Type:

bool | None

log_ignored_actions

Message actions that should not be logged

Type:

collections.abc.Iterable[str] | None

send_authentication_message

Whether to send auth status after connection

Type:

bool | None

INCOMING_MESSAGE_SCHEMA

Pydantic model class for validating incoming messages

Type:

type[chanx.messages.base.BaseIncomingMessage]

OUTGOING_GROUP_MESSAGE_SCHEMA

Pydantic model class for validating group broadcast messages, required when using send_group_message with kind="message"

Type:

type[chanx.messages.base.BaseOutgoingGroupMessage]

async add_groups() → None

Add the consumer to channel groups.

Retrieves groups from build_groups() and adds this consumer to each channel group for broadcast messaging.

authenticator_class

alias of ChanxWebsocketAuthenticator

async build_groups() → list[str]

Build list of channel groups to join.

Subclasses should override this method to define which groups the consumer should join based on authentication results.

Returns:

Iterable of group names to join

async post_authentication() → None

Hook for additional actions after successful authentication.

Subclasses can override this method to perform custom actions after a successful authentication.

async receive_json(content: dict[str, Any], **kwargs: Any) → None

Receive and process JSON data from WebSocket.

Logs messages, assigns ID, and creates task for async processing.

Parameters:

• content -- The JSON content received from the client

• **kwargs -- Additional keyword arguments

abstract async receive_message(message: BaseMessage, **kwargs: Any) → None

Process a validated received message.

Must be implemented by subclasses to handle messages after validation.

Parameters:

• message -- The validated message object

• **kwargs -- Additional keyword arguments

async send_group_member(event: GroupMemberEvent) → None

Handle incoming group message and relay to client.

Processes group messages from the channel layer, adds metadata like is_mine and is_current, and forwards to the client socket. This method is called by the Channels system when a message is sent to a group this consumer is part of.

The method adds two metadata fields to all messages:

• is_mine: True if the message originated from the current user

• is_current: True if the message originated from this channel

If the message is from the current channel and exclude_current is True, the message is not relayed to avoid echo effects. For message-type events, the content is wrapped in the OUTGOING_GROUP_MESSAGE_SCHEMA, while JSON-type events are sent directly. If configured, a GroupCompleteMessage is sent after successful processing.

Parameters:

event -- Group member event data containing the content, kind, source channel, user ID, and control flags

async send_group_message(message: BaseMessage, groups: list[str] | None = None, *, kind: Literal['json', 'message'] = 'message', exclude_current: bool = True) → None

Send a BaseMessage object to one or more channel groups.

Broadcasts a message to all consumers in the specified groups. This is useful for implementing pub/sub patterns where messages need to be distributed to multiple connected clients.

Important

When using kind="message" (the default), your consumer class must define OUTGOING_GROUP_MESSAGE_SCHEMA to properly validate and wrap the message. This schema ensures that group messages follow the expected structure and contain the required metadata. If not defined, use kind="json" instead.

Parameters:

• message -- Message object to send to the groups

• groups -- Group names to send to (defaults to self.groups)

• kind --

  Format to send the message as:

  • "message": Validated and wrapped via OUTGOING_GROUP_MESSAGE_SCHEMA (default)

  • "json": Sent as raw JSON without validation or wrapping

• exclude_current -- Whether to exclude the sending consumer from receiving the broadcast (prevents echo effects)

async send_json(content: dict[str, Any], close: bool = False) → None

Send JSON data to the WebSocket client.

Sends data and optionally logs it.

Parameters:

• content -- The JSON content to send

• close -- Whether to close the connection after sending

async send_message(message: BaseMessage) → None

Send a Message object to the WebSocket client.

Serializes the message and sends it as JSON.

Parameters:

message -- The Message object to send

async send_to_groups(content: dict[str, Any], groups: list[str] | None = None, *, exclude_current: bool = True, kind: Literal['json', 'message'] = 'json') → None

Send content to one or more channel groups.

Low-level method to broadcast dictionary content to channel groups. For sending BaseMessage objects, prefer using send_group_message() instead.

Parameters:

• content -- Dictionary content to send to the groups

• groups -- Group names to send to (defaults to self.groups)

• exclude_current -- Whether to exclude the sending consumer from receiving the broadcast (prevents echo effects)

• kind --

  Type of message to send:

  • "json": Send as raw JSON directly to clients (default)

  • "message": Process through OUTGOING_GROUP_MESSAGE_SCHEMA validation (requires consumer to define this schema)

async websocket_connect(message: WebSocketConnectEvent) → None

Handle WebSocket connection request with authentication.

Accepts the connection, authenticates the user, and either adds the user to appropriate groups or closes the connection.

Parameters:

message -- The connection message from Channels

async websocket_disconnect(message: WebSocketDisconnectEvent) → None

Handle WebSocket disconnection.

Cleans up context variables and logs the disconnection.

Parameters:

message -- The disconnection message from Channels

Authenticator

class chanx.generic.authenticator.ChanxWebsocketAuthenticator

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]] | None

permission_classes

DRF permission classes for connection authorization

Type:

collections.abc.Sequence[type[rest_framework.permissions.BasePermission] | rest_framework.permissions.OperandHolder | rest_framework.permissions.SingleOperandHolder] | None

queryset

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

Type:

Literal[True] | django.db.models.query.QuerySet | django.db.models.manager.Manager

auth_method

HTTP verb to emulate for authentication

Type:

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

class chanx.generic.authenticator.AuthenticationResult(is_authenticated: bool, status_code: int, status_text: str, data: dict[str, ~typing.Any] = <factory>, user: ~django.contrib.auth.base_user.AbstractBaseUser | ~django.contrib.auth.models.AnonymousUser | None = None, obj: ~django.db.models.base.Model | None = None)

Result of websocket authentication.

is_authenticated

Whether authentication was successful

Type:

bool

status_code

HTTP status code representing auth result

Type:

int

status_text

Text description of status

Type:

str

data

Additional data related to authentication result

Type:

dict[str, Any]

user

Authenticated user object if successful

Type:

django.contrib.auth.base_user.AbstractBaseUser | django.contrib.auth.models.AnonymousUser | None

obj

Related model object if requested

Type:

django.db.models.base.Model | None

class chanx.generic.authenticator.ChanxSerializer(*args, **kwargs)

Bases: Serializer

Base serializer for Chanx authentication.

class chanx.generic.authenticator.ChanxAuthView(**kwargs)

Base authentication view for Chanx websockets.

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

delete(request: Request, *args: Any, **kwargs: Any) → Response

Stub delete method

get(request: Request, *args: Any, **kwargs: Any) → Response

Stub get method

get_response(request: Request) → Response

Get standard response with object if required.

Parameters:

request -- The HTTP request object

Returns:

Response with OK detail

patch(request: Request, *args: Any, **kwargs: Any) → Response

Stub patch method

post(request: Request, *args: Any, **kwargs: Any) → Response

Stub post method

put(request: Request, *args: Any, **kwargs: Any) → Response

Stub put method

serializer_class

alias of ChanxSerializer