FastAPI & ASGI Frameworks Integration

Chanx provides integration for FastAPI and other ASGI-based frameworks through the fast-channels module, including WebSocket consumers, AsyncAPI documentation views, and testing utilities. Works with any ASGI framework including FastAPI, Starlette, Litestar, and more.

WebSocket Consumer

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

FastAPI fast-channels WebSocket consumer with Chanx enhanced features.

Combines Chanx's automatic message handling, type validation, and group broadcasting capabilities with fast-channels' WebSocket consumer for FastAPI.

Features from Chanx mixin:

Automatic message type discovery from @ws_handler decorators
Type-safe message validation using Pydantic discriminated unions
Group broadcasting and channel event system
Comprehensive error handling and logging
Optional authentication support

Features from fast-channels:

FastAPI ASGI integration
fast-channels channel layer support (Redis, in-memory, etc.)
WebSocket lifecycle management
High-performance async operation

get_channel_layer() → BaseChannelLayer | None: Returns a channel layer by alias.

Views

async chanx.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.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.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.

Testing

class chanx.fast_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]])

Fast-channels WebSocket communicator for testing Chanx consumers.

Combines Chanx testing mixin features with fast-channels' WebSocket communicator, providing comprehensive testing capabilities for FastAPI applications:

Chanx features (from WebsocketCommunicatorMixin):

Structured message sending/receiving with BaseMessage objects
Automatic message collection until completion signals
Message validation using consumer's type adapters
Async context manager support for automatic cleanup
send_message(): Send BaseMessage objects directly
receive_all_json(): Collect all messages until timeout
receive_all_messages(): Collect and validate messages until stop action

FastAPI fast-channels features:

Full compatibility with FastAPI ASGI applications
fast-channels channel layer support
High-performance async operation

Configuration

For ASGI-based frameworks, configuration is done via class attributes on your consumer:

from chanx.fast_channels.websocket import AsyncJsonWebsocketConsumer

class BaseConsumer(AsyncJsonWebsocketConsumer):
    # Message handling
    camelize = False
    discriminator_field = "action"
    send_completion = False
    send_message_immediately = True

    # Logging
    log_websocket_message = False
    log_ignored_actions = []

    # Channel layer
    channel_layer_alias = "default"

Configuration Options

Attribute	Default	Description
`discriminator_field`	`"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
`send_message_immediately`	`True`	Whether to send message immediately rather than wait to the end of processing
`camelize`	`False`	Whether to convert message keys to camelCase format using pyhumps
`log_websocket_message`	`False`	Whether to log WebSocket messages (both received and sent)
`log_ignored_actions`	`[]`	List of message actions that should not be logged
`channel_layer_alias`	`"default"`	Channel layer alias to use for broadcasting and events