Extensions

Chanx provides framework-specific extensions for Django Channels and FastAPI.

Django Channels Extension

The Channels extension provides seamless integration with Django Channels, including views for AsyncAPI documentation and testing utilities.

Views

class chanx.ext.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.ext.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.ext.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

Authenticator

Django REST Framework integration for WebSocket authentication.

class chanx.ext.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.ext.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.ext.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[AsyncJsonWebsocketConsumer[Any]])

Django-enhanced WebSocket communicator for Chanx testing.

Extends the core WebsocketCommunicator with Django-specific features:

Django authentication message handling with wait_for_auth()
Integration with Django settings (SEND_AUTHENTICATION_MESSAGE, CAMELIZE)
REST framework status code validation with assert_authenticated_status_ok()
Automatic camelCase/snake_case conversion based on Django settings

Django-specific methods: - wait_for_auth(): Handle Django authentication messages - assert_authenticated_status_ok(): Validate DRF authentication status

Inherits all core functionality from WebsocketCommunicator including structured message handling, completion signals, and message validation.

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 assert_closed() → None: Asserts that the WebSocket has been closed.

async connect(timeout: float = 1) → tuple[bool, int | str | None]

Connects to the WebSocket and tracks connection state.

Parameters:: timeout -- Maximum time to wait for connection (in seconds)
Returns:: Tuple of (connected, status_code)

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.ext.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.ext.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 extension 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.ext.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.

FastAPI Extension

The FastAPI extension provides integration utilities for FastAPI applications.

Views

async chanx.ext.fast_channels.views.asyncapi_docs(request: Request, app: Starlette | FastAPI, config: AsyncAPIConfig | None = None) → HTMLResponse: Simple AsyncAPI docs endpoint that works with FastAPI, Starlette, etc. Follows the same template structure as Django's channels extension.

async chanx.ext.fast_channels.views.asyncapi_spec_json(request: Request, app: Starlette | FastAPI, config: AsyncAPIConfig | None = None) → JSONResponse

Simple JSON spec endpoint that works with FastAPI, Starlette, etc.

Usage with FastAPI:: @app.get("/asyncapi.json") async def get_spec():

return await asyncapi_spec_json(app=app)
Usage with config:: @app.get("/asyncapi.json") async def get_spec():

return await asyncapi_spec_json(app=app, config={"title": "My API"})
Usage with Starlette:: Route("/asyncapi.json", asyncapi_spec_json)

async chanx.ext.fast_channels.views.asyncapi_spec_yaml(request: Request, app: Starlette | FastAPI, config: AsyncAPIConfig | None = None) → Response: Simple YAML spec endpoint that works with FastAPI, Starlette, etc.