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.

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 authentication

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 message validation

Type:

type[chanx.messages.base.BaseIncomingMessage]

OUTGOING_GROUP_MESSAGE_SCHEMA

Pydantic model class for group messages

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() → Iterable[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.

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, user ownership is tracked. 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, *, exclude_current: bool = True) → None

Send a BaseMessage object to one or more groups.

Parameters:

• message – Message object to send

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

• exclude_current – Whether to exclude the sending consumer

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.

Parameters:

• content – Content to send

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

• exclude_current – Whether to exclude the sending consumer

• kind – Type of message (‘json’ or ‘message’)

async websocket_connect(message: dict[str, Any]) → 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: dict[str, Any]) → 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: ExtendedRequest, *args: Any, **kwargs: Any) → Response

Stub delete method

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

Stub get method

get_response(request: ExtendedRequest) → Response

Get standard response with object if required.

Parameters:

request – The HTTP request object

Returns:

Response with OK status and object if needed

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

Stub patch method

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

Stub post method

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

Stub put method

serializer_class

alias of ChanxSerializer