Messages and schemas¶

from litestar_asyncapi.spec import Message, MessageExample, MessageTrait, MultiFormatSchema, Schema

Generated payload and header schemas are JSON Schema Draft07 dictionaries. Use Schema or MultiFormatSchema to write one by hand.

Messages¶

`litestar_asyncapi.spec.Message` ¶

Bases: BaseSchemaObject

AsyncAPI Message object.

`name: str | None = None` ¶

Machine-friendly message name.

`deprecated: bool | None = None` ¶

Whether the message is deprecated.

`title: str | None = None` ¶

Human-friendly message title.

`summary: str | None = None` ¶

Short summary of the message.

`description: str | None = None` ¶

Longer description of the message; CommonMark is allowed.

`headers: Schema | Reference | MultiFormatSchema | dict[str, Any] | bool | None = None` ¶

Schema of the message headers.

`payload: Schema | Reference | MultiFormatSchema | dict[str, Any] | bool | None = None` ¶

Schema of the message payload.

`correlation_id: CorrelationId | Reference | None = None` ¶

Location of the correlation id in the message.

`content_type: str | None = None` ¶

Media type of the payload; the document's default content type applies when omitted.

`tags: list[Tag | Reference] | None = None` ¶

Tags for logical grouping.

`external_docs: ExternalDocumentation | Reference | None = None` ¶

Additional external documentation.

`traits: list[MessageTrait | Reference] | None = None` ¶

Traits applied to the message.

`bindings: dict[str, Any] | Reference | None = None` ¶

Protocol-specific message bindings.

`examples: list[MessageExample] | None = None` ¶

Example messages.

`extensions: dict[str, Any] = field(default_factory=dict)` ¶

Specification extensions; every key must start with x-.

`litestar_asyncapi.spec.MessageTrait` ¶

Bases: BaseSchemaObject

Reusable message properties, excluding payload and traits.

`name: str | None = None` ¶

Machine-friendly message name.

`headers: Schema | Reference | MultiFormatSchema | dict[str, Any] | bool | None = None` ¶

Schema of the message headers.

`correlation_id: CorrelationId | Reference | None = None` ¶

Location of the correlation id in the message.

`content_type: str | None = None` ¶

Media type of the payload.

`examples: list[MessageExample] | None = None` ¶

Example messages.

`extensions: dict[str, Any] = field(default_factory=dict)` ¶

Specification extensions; every key must start with x-.

`deprecated: bool | None = None` ¶

Whether the message is deprecated.

`title: str | None = None` ¶

Human-friendly message title.

`summary: str | None = None` ¶

Short summary of the message.

`description: str | None = None` ¶

Longer description of the message; CommonMark is allowed.

`tags: list[Tag | Reference] | None = None` ¶

Tags for logical grouping.

`external_docs: ExternalDocumentation | Reference | None = None` ¶

Additional external documentation.

`bindings: dict[str, Any] | Reference | None = None` ¶

Protocol-specific message bindings.

`litestar_asyncapi.spec.MessageExample` ¶

Bases: BaseSchemaObject

An example message with independently optional headers and payload.

`headers: dict[str, Any] | None = None` ¶

Example header values.

`payload: Any = field(default=UNSET, metadata={'nullable': True})` ¶

Example payload; None is emitted as an explicit null and an unset value is omitted.

`name: str | None = None` ¶

Machine-friendly example name.

`summary: str | None = None` ¶

Short summary of the example.

`litestar_asyncapi.spec.CorrelationId` ¶

Bases: BaseSchemaObject

Defines how to extract a correlation id from a message.

`location: str` ¶

Runtime expression for the correlation id, such as $message.header#/correlationId.

`description: str | None = None` ¶

Description of the identifier; CommonMark is allowed.

`extensions: dict[str, Any] = field(default_factory=dict)` ¶

Specification extensions; every key must start with x-.

Schemas¶

`litestar_asyncapi.spec.Schema` ¶

Bases: BaseSchemaObject

A JSON Schema (Draft 07 aligned) used in AsyncAPI message payloads and headers.

`to_schema() -> dict[str, Any]` ¶

Serialize the schema with Draft07 tuple keywords.

prefix_items becomes a positional items array and items becomes additionalItems, so no prefixItems reaches the document.

Returns:

Type	Description
`dict[str, Any]`	A dictionary representation of the schema.

Raises:

Type	Description
`ValueError`	If `items` and `additional_items` are both set beside `prefix_items`, or `items` is a list beside `prefix_items`.

`litestar_asyncapi.spec.MultiFormatSchema` ¶

Bases: BaseSchemaObject

A schema in an explicitly declared non-native format.

`schema_format: str` ¶

Schema format identifier, such as application/vnd.apache.avro;version=1.9.0.

`schema: Any = field(metadata={'nullable': True})` ¶

Schema in the declared format.

`extensions: dict[str, Any] = field(default_factory=dict)` ¶

Specification extensions; every key must start with x-.

`litestar_asyncapi.spec.SchemaType` ¶

Bases: StrEnum

JSON Schema base types (Draft 07 compatible).

`ARRAY = 'array'` ¶

`BOOLEAN = 'boolean'` ¶

`INTEGER = 'integer'` ¶

`NULL = 'null'` ¶

`NUMBER = 'number'` ¶

`OBJECT = 'object'` ¶

`STRING = 'string'` ¶

`litestar_asyncapi.spec.SchemaFormat` ¶

Bases: StrEnum

Common JSON Schema formats used by AsyncAPI payloads.

Messages and schemas¶

Messages¶

litestar_asyncapi.spec.Message ¶

name: str | None = None ¶

deprecated: bool | None = None ¶

title: str | None = None ¶

summary: str | None = None ¶

description: str | None = None ¶

headers: Schema | Reference | MultiFormatSchema | dict[str, Any] | bool | None = None ¶

payload: Schema | Reference | MultiFormatSchema | dict[str, Any] | bool | None = None ¶

correlation_id: CorrelationId | Reference | None = None ¶

content_type: str | None = None ¶

tags: list[Tag | Reference] | None = None ¶

external_docs: ExternalDocumentation | Reference | None = None ¶

traits: list[MessageTrait | Reference] | None = None ¶

bindings: dict[str, Any] | Reference | None = None ¶

examples: list[MessageExample] | None = None ¶

extensions: dict[str, Any] = field(default_factory=dict) ¶

litestar_asyncapi.spec.MessageTrait ¶

name: str | None = None ¶

headers: Schema | Reference | MultiFormatSchema | dict[str, Any] | bool | None = None ¶

correlation_id: CorrelationId | Reference | None = None ¶

content_type: str | None = None ¶

examples: list[MessageExample] | None = None ¶

extensions: dict[str, Any] = field(default_factory=dict) ¶

deprecated: bool | None = None ¶

title: str | None = None ¶

summary: str | None = None ¶

description: str | None = None ¶

tags: list[Tag | Reference] | None = None ¶

external_docs: ExternalDocumentation | Reference | None = None ¶

bindings: dict[str, Any] | Reference | None = None ¶

litestar_asyncapi.spec.MessageExample ¶

headers: dict[str, Any] | None = None ¶

payload: Any = field(default=UNSET, metadata={'nullable': True}) ¶

name: str | None = None ¶

summary: str | None = None ¶

litestar_asyncapi.spec.CorrelationId ¶

location: str ¶

description: str | None = None ¶

extensions: dict[str, Any] = field(default_factory=dict) ¶

Schemas¶

litestar_asyncapi.spec.Schema ¶

to_schema() -> dict[str, Any] ¶

litestar_asyncapi.spec.MultiFormatSchema ¶

schema_format: str ¶

schema: Any = field(metadata={'nullable': True}) ¶

extensions: dict[str, Any] = field(default_factory=dict) ¶

litestar_asyncapi.spec.SchemaType ¶

ARRAY = 'array' ¶

BOOLEAN = 'boolean' ¶

INTEGER = 'integer' ¶

NULL = 'null' ¶

NUMBER = 'number' ¶

OBJECT = 'object' ¶

STRING = 'string' ¶

litestar_asyncapi.spec.SchemaFormat ¶

DATE = 'date' ¶

DATE_TIME = 'date-time' ¶

DURATION = 'duration' ¶

EMAIL = 'email' ¶

HOSTNAME = 'hostname' ¶

IPV4 = 'ipv4' ¶

IPV6 = 'ipv6' ¶

REGEX = 'regex' ¶

URI = 'uri' ¶

UUID = 'uuid' ¶

`litestar_asyncapi.spec.Message` ¶

`name: str | None = None` ¶

`deprecated: bool | None = None` ¶

`title: str | None = None` ¶

`summary: str | None = None` ¶

`description: str | None = None` ¶

`headers: Schema | Reference | MultiFormatSchema | dict[str, Any] | bool | None = None` ¶

`payload: Schema | Reference | MultiFormatSchema | dict[str, Any] | bool | None = None` ¶

`correlation_id: CorrelationId | Reference | None = None` ¶

`content_type: str | None = None` ¶

`tags: list[Tag | Reference] | None = None` ¶

`external_docs: ExternalDocumentation | Reference | None = None` ¶

`traits: list[MessageTrait | Reference] | None = None` ¶

`bindings: dict[str, Any] | Reference | None = None` ¶

`examples: list[MessageExample] | None = None` ¶

`extensions: dict[str, Any] = field(default_factory=dict)` ¶

`litestar_asyncapi.spec.MessageTrait` ¶

`name: str | None = None` ¶

`headers: Schema | Reference | MultiFormatSchema | dict[str, Any] | bool | None = None` ¶

`correlation_id: CorrelationId | Reference | None = None` ¶

`content_type: str | None = None` ¶

`examples: list[MessageExample] | None = None` ¶

`extensions: dict[str, Any] = field(default_factory=dict)` ¶

`deprecated: bool | None = None` ¶

`title: str | None = None` ¶

`summary: str | None = None` ¶

`description: str | None = None` ¶

`tags: list[Tag | Reference] | None = None` ¶

`external_docs: ExternalDocumentation | Reference | None = None` ¶

`bindings: dict[str, Any] | Reference | None = None` ¶

`litestar_asyncapi.spec.MessageExample` ¶

`headers: dict[str, Any] | None = None` ¶

`payload: Any = field(default=UNSET, metadata={'nullable': True})` ¶

`name: str | None = None` ¶

`summary: str | None = None` ¶

`litestar_asyncapi.spec.CorrelationId` ¶

`location: str` ¶

`description: str | None = None` ¶

`extensions: dict[str, Any] = field(default_factory=dict)` ¶

`litestar_asyncapi.spec.Schema` ¶

`to_schema() -> dict[str, Any]` ¶

`litestar_asyncapi.spec.MultiFormatSchema` ¶

`schema_format: str` ¶

`schema: Any = field(metadata={'nullable': True})` ¶

`extensions: dict[str, Any] = field(default_factory=dict)` ¶

`litestar_asyncapi.spec.SchemaType` ¶

`ARRAY = 'array'` ¶

`BOOLEAN = 'boolean'` ¶

`INTEGER = 'integer'` ¶

`NULL = 'null'` ¶

`NUMBER = 'number'` ¶

`OBJECT = 'object'` ¶

`STRING = 'string'` ¶

`litestar_asyncapi.spec.SchemaFormat` ¶

`DATE = 'date'` ¶

`DATE_TIME = 'date-time'` ¶

`DURATION = 'duration'` ¶

`EMAIL = 'email'` ¶

`HOSTNAME = 'hostname'` ¶

`IPV4 = 'ipv4'` ¶

`IPV6 = 'ipv6'` ¶

`REGEX = 'regex'` ¶

`URI = 'uri'` ¶

`UUID = 'uuid'` ¶