AsyncAPI Module
The AsyncAPI module provides automatic documentation generation for Chanx WebSocket consumers, creating AsyncAPI 3.0 specifications from decorated consumers.
Generator
- class chanx.asyncapi.generator.AsyncAPIGenerator(routes: list[RouteInfo], title: str | None = 'AsyncAPI Documentation', version: str | None = '1.0.0', description: str | None = None, server_url: str | None = 'localhost:8000', server_protocol: str | None = 'ws')
Generates AsyncAPI 3.0 specifications from Chanx WebSocket routes.
This class analyzes WebSocket consumer routes and their decorated handlers to automatically generate comprehensive AsyncAPI documentation including channels, operations, messages, and schemas.
- build_channels() dict[str, dict[str, Any]]
Build AsyncAPI channels from WebSocket routes.
Analyzes each route and its consumer to create channel definitions including parameters, messages, and metadata from @channel decorators.
- Returns:
Dictionary of channel name to channel specification
- build_operations() None
Build AsyncAPI operations from WebSocket and event handlers.
Scans all consumers for @ws_handler and @event_handler decorated methods and creates corresponding send/receive operations with proper message references and reply definitions.
- build_output(channel_name: str, output_type: type[BaseMessage]) dict[str, Any]
Build an output message reference for operation responses.
- Parameters:
channel_name -- The channel name containing the message
output_type -- The BaseMessage subclass for the output
- Returns:
Message reference dictionary for AsyncAPI specification
- generate() dict[str, Any]
Generate the complete AsyncAPI 3.0 specification.
Builds channels and operations from the provided routes, then constructs the final AsyncAPI document with all components.
- Returns:
Complete AsyncAPI 3.0 specification as a dictionary
- get_channel_messages(consumer: type[AsyncJsonWebsocketConsumer]) dict[str, dict[str, Any]]
Extract message definitions for a channel from its consumer.
- Parameters:
consumer -- The WebSocket consumer class
- Returns:
Dictionary mapping message names to message references
Constants
- chanx.asyncapi.constants.DEFAULT_ASYNCAPI_TITLE = 'AsyncAPI Documentation'
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.
- chanx.asyncapi.constants.DEFAULT_ASYNCAPI_VERSION = '1.0.0'
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.
- chanx.asyncapi.constants.DEFAULT_SERVER_URL = 'localhost:8000'
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.
- chanx.asyncapi.constants.DEFAULT_SERVER_PROTOCOL = 'ws'
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.
Type Definitions
- class chanx.asyncapi.type_defs.AsyncAPIDocument(*, asyncapi: str = '3.0.0', info: InfoObject, servers: dict[str, ServerObject] | None = None, channels: dict[str, ChannelObject] | None = None, operations: dict[str, OperationObject] | None = None, components: ComponentsObject | None = None, tags: list[TagObject] | None = None, externalDocs: ExternalDocumentationObject | None = None)
Root AsyncAPI 3.0 document containing the complete API specification.
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class chanx.asyncapi.type_defs.InfoObject(*, title: str, version: str, description: str | None = None, termsOfService: str | None = None, contact: ContactObject | None = None, license: LicenseObject | None = None, tags: list[TagObject] | None = None, externalDocs: ExternalDocumentationObject | None = None)
AsyncAPI Info Object containing API metadata.
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class chanx.asyncapi.type_defs.ServerObject(*, url: str | None = None, host: str | None = None, protocol: str | None = None, protocolVersion: str | None = None, pathname: str | None = None, description: str | None = None, title: str | None = None, summary: str | None = None, security: list[dict[str, list[str]]] | None = None, tags: list[TagObject] | None = None, externalDocs: ExternalDocumentationObject | None = None, bindings: dict[str, dict[str, Any]] | None = None, variables: dict[str, ServerVariableObject] | None = None)
AsyncAPI Server Object describing a server where the API is hosted.
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class chanx.asyncapi.type_defs.ChannelObject(*, address: str | None = None, title: str | None = None, summary: str | None = None, description: str | None = None, servers: list[str] | None = None, parameters: dict[str, ParameterObject] | None = None, messages: dict[str, MessageObject | dict[str, Any]] | None = None, bindings: dict[str, dict[str, Any]] | None = None, subscribe: OperationObject | None = None, publish: OperationObject | None = None, tags: list[TagObject] | None = None, externalDocs: ExternalDocumentationObject | None = None)
AsyncAPI Channel Object describing a communication channel.
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class chanx.asyncapi.type_defs.OperationObject(*, action: str | None = None, channel: dict[str, Any] | None = None, title: str | None = None, summary: str | None = None, description: str | None = None, security: list[dict[str, list[str]]] | None = None, tags: list[TagObject] | None = None, externalDocs: ExternalDocumentationObject | None = None, bindings: dict[str, dict[str, Any]] | None = None, traits: list[OperationTraitObject | dict[str, Any]] | None = None, messages: list[MessageObject | dict[str, Any]] | None = None, reply: OperationReplyObject | None = None)
AsyncAPI Operation Object describing send/receive operations on channels.
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class chanx.asyncapi.type_defs.MessageObject(*, ref: str | None = None, name: str | None = None, title: str | None = None, summary: str | None = None, description: str | None = None, contentType: str | None = None, schemaFormat: str | None = None, headers: SchemaObject | None = None, payload: SchemaObject | None = None, correlationId: CorrelationIdObject | None = None, tags: list[TagObject] | None = None, externalDocs: ExternalDocumentationObject | None = None, bindings: dict[str, dict[str, Any]] | None = None, traits: list[MessageTraitObject | dict[str, Any]] | None = None)
AsyncAPI Message Object describing a message payload and metadata.
- class Config
Pydantic configuration for MessageObject to support alias field names.
- model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class chanx.asyncapi.type_defs.SchemaObject(*, ref: str | None = None, title: str | None = None, description: str | None = None, type: str | list[str] | None = None, format: str | None = None, default: Any | None = None, enum: list[Any] | None = None, const: Any | None = None, multipleOf: int | float | None = None, maximum: int | float | None = None, exclusiveMaximum: int | float | None = None, minimum: int | float | None = None, exclusiveMinimum: int | float | None = None, maxLength: int | None = None, minLength: int | None = None, pattern: str | None = None, items: SchemaObject | list[SchemaObject] | None = None, maxItems: int | None = None, minItems: int | None = None, uniqueItems: bool | None = None, contains: SchemaObject | None = None, prefixItems: list[SchemaObject] | None = None, maxProperties: int | None = None, minProperties: int | None = None, required: list[str] | None = None, properties: dict[str, SchemaObject] | None = None, patternProperties: dict[str, SchemaObject] | None = None, additionalProperties: SchemaObject | bool | None = None, propertyNames: SchemaObject | None = None, unevaluatedProperties: SchemaObject | bool | None = None, allOf: list[SchemaObject] | None = None, anyOf: list[SchemaObject] | None = None, oneOf: list[SchemaObject] | None = None, not_: SchemaObject | None = None, if_: SchemaObject | None = None, then: SchemaObject | None = None, else_: SchemaObject | None = None, dependentSchemas: dict[str, SchemaObject] | None = None, deprecated: bool | None = None, examples: list[Any] | None = None)
AsyncAPI Schema Object representing JSON Schema subset for message payloads.
- class Config
Pydantic configuration for SchemaObject to support alias field names.
- model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].