Utils Module

ASGI Utilities

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

Asyncio Utilities

chanx.utils.asyncio.create_task(coro: Coroutine[Any, Any, T], background_tasks: set[Task[Any]] | None = None, name: str | None = None, context: Context | None = None) → Task[T]

Creates an asyncio task with proper cleanup and tracking.

This function creates a task from a coroutine, wrapping it with error handling and adding it to a set of background tasks for tracking. The task automatically removes itself from the tracking set when completed.

Parameters:

  • coro – The coroutine to convert into a task

  • background_tasks – Set to track the task (uses global set if None)

  • name – Optional name for the task

  • context – Optional context for the task

Returns:

The created asyncio task

Logging Utilities

chanx.utils.logging.logger = structlog.get_logger("chanx")

Instantiates a bound logger on first usage.

Takes both configuration and instantiation parameters into account.

The only points where a bound logger changes state are bind(), unbind(), and new() and that return the actual BoundLogger.

If and only if configuration says so, that actual bound logger is cached on first usage.

Changed in version 0.4.0: Added support for logger_factory_args.

WebSocket Utilities

class chanx.utils.websocket.RouteInfo(path: str, handler: Any, base_url: str, path_params: dict[str, str] | None = None)

WebSocket route information.

path

The URL path pattern for the WebSocket route.

Type:

str

handler

The consumer or handler function for this route.

Type:

Any

base_url

The base WebSocket URL (e.g., ws://domain.com).

Type:

str

path_params

Dictionary of path parameters with their regex patterns.

Type:

dict[str, str] | None

chanx.utils.websocket.get_websocket_routes(request: HttpRequest | None = None) → list[RouteInfo]

Discover all WebSocket routes from the ASGI application.

This function traverses the Django Channels routing configuration to discover all available WebSocket endpoints and their handlers. The discovered routes can be used for various purposes like documentation, testing, or validation.

Parameters:

request – The HTTP request object, used to determine the current domain. If None, defaults to localhost:8000.

Returns:

A list of RouteInfo objects containing path, handler, and base_url for each WebSocket route.

chanx.utils.websocket.transform_routes(routes: list[RouteInfo], transform_fn: Callable[[RouteInfo], T]) → list[T]

Apply a transformation function to all discovered routes.

This helper function applies a custom transformation to each route, allowing for flexible processing of route information.

Parameters:

  • routes – List of RouteInfo dataclass instances

  • transform_fn – Function to transform each route into the desired format

Returns:

List of transformed route information with the type determined by transform_fn