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, typed channel events, group messaging, logging, and error handling. Subclasses must implement receive_message and specify the incoming message type as a generic parameter.

For typed channel events, subclasses can define a union type of channel events and use the Event generic parameter to enable type-safe channel event handling. Override the receive_event() method to process events sent via send_channel_event() or asend_channel_event(). Events are automatically validated against the Event type before being passed to your handler method.

For group messaging functionality, subclasses should also define the outgoing group message type using the OG generic parameter to enable proper validation and handling of group message broadcasts.

Generic Parameters:

IC: Incoming message type (required) - Union of BaseMessage subclasses Event: Channel event type (optional) - Union of BaseChannelEvent subclasses or None OG: Outgoing group message type (optional) - BaseGroupMessage subclass or None M: Model type for object-level permissions (optional) - Model subclass or None

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

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.

async classmethod asend_channel_event(group_name: str, event: Event) → None

Send a typed channel event to a channel group.

This is a class method that provides a type-safe way to send events through the channel layer to consumers. It can be called from tasks, views, or other places where you don't have a consumer instance. The event type is constrained by the consumer's Event generic parameter.

Parameters:

• group_name -- Group name to send the event to

• event -- The typed event to send (must match the consumer's Event type)

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 handle_channel_event(event_payload: EventPayload) → None

Internal dispatcher for typed channel events with completion signal.

This method is called by the channel layer when an event is sent to a group this consumer belongs to. It validates the event data and forwards it to the receive_event method.

Parameters:

event_payload -- The message from the channel layer containing event data

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_event(event: Event) → None

Process typed channel events received through the channel layer.

This method is called when a channel event is sent to a group this consumer belongs to. Override this method to handle events sent via send_channel_event() or asend_channel_event().

Channel events provide a way to send typed messages to consumers from outside the WebSocket connection (e.g., from Django views, tasks, or other consumers). Use pattern matching to handle different event types based on your Event generic parameter.

Parameters:

event -- The validated event object (typed based on Event generic parameter)

Note

This method is only called if your consumer defines the Event generic parameter. If Event is None, channel events are not supported.

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: IC, **kwargs: Any) → None

Process a validated received message.

Must be implemented by subclasses to handle messages after validation. The message parameter is automatically typed based on the IC generic parameter.

Parameters:

• message -- The validated message object (typed as IC)

• **kwargs -- Additional keyword arguments

classmethod send_channel_event(group_name: str, event: Event) → None

Synchronous version of asend_channel_event for use in Django tasks/views.

This method provides the same functionality as asend_channel_event but can be called from synchronous code like Django tasks, views, or signals.

Parameters:

• group_name -- Group name to send to

• event -- The typed event to send (constrained by the consumer's Event type)

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 the OG generic parameter (outgoing group message type) to properly validate and handle the message. 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 via OG generic parameter (default)

  • "json": Sent as raw JSON without validation

• 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