Skip to content

Bindings and security

from litestar_asyncapi.spec import SecurityScheme, SecuritySchemeType
from litestar_asyncapi.spec.bindings import KafkaChannelBinding, WebSocketChannelBinding

Security schemes describe how clients authenticate; declaring one does not enforce it. Bindings add protocol-specific details to servers, channels, operations and messages. Use a binding model as a value in a bindings mapping keyed by protocol, such as {"kafka": KafkaChannelBinding(partitions=3)}.

Security

litestar_asyncapi.spec.SecurityScheme

Bases: BaseSchemaObject

AsyncAPI Security Scheme object.

type: SecuritySchemeType

Type of the security scheme.

description: str | None = None

Description of the scheme; CommonMark is allowed.

name: str | None = None

Name of the header, query or cookie parameter for httpApiKey schemes.

in_: str | None = field(default=None, metadata={'alias': 'in'})

Location of the key, serialized as in: user or password for apiKey, query, header or cookie for httpApiKey.

scopes: list[str] | None = None

Scopes required by the operation or server.

scheme: str | None = None

HTTP authorization scheme for http schemes, such as bearer.

bearer_format: str | None = None

Format hint for bearer tokens, such as JWT.

open_id_connect_url: str | None = None

OpenID Connect discovery URL for openIdConnect schemes.

flows: dict[str, Any] | None = None

OAuth2 flow configuration for oauth2 schemes.

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

Specification extensions; every key must start with x-.

litestar_asyncapi.spec.SecuritySchemeType

Bases: StrEnum

AsyncAPI Security Scheme types.

API_KEY = 'apiKey'

HTTP = 'http'

OAUTH2 = 'oauth2'

OPEN_ID_CONNECT = 'openIdConnect'

USER_PASSWORD = 'userPassword'

X509 = 'X509'

SYMMETRIC_ENCRYPTION = 'symmetricEncryption'

ASYMMETRIC_ENCRYPTION = 'asymmetricEncryption'

HTTP_API_KEY = 'httpApiKey'

SASL_PLAIN = 'plain'

SASL_SCRAM_SHA256 = 'scramSha256'

SASL_SCRAM_SHA512 = 'scramSha512'

SASL_GSSAPI = 'gssapi'

Bindings

litestar_asyncapi.spec.bindings.Binding

Bases: BaseSchemaObject

Base class for protocol binding objects.

Bindings commonly have a bindingVersion field and may include vendor extensions.

binding_version: str | None = None

Version of the binding specification, serialized as bindingVersion.

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

Specification extensions; every key must start with x-.

WebSocket

litestar_asyncapi.spec.bindings.WebSocketServerBinding

Bases: Binding

WebSocket server binding, which defines no protocol-specific fields.

litestar_asyncapi.spec.bindings.WebSocketChannelBinding

Bases: Binding

WebSocket channel binding describing the opening handshake.

method: str | None = None

HTTP method of the opening handshake: GET or POST.

query: dict[str, Any] | None = None

Schema of the query parameters of the opening handshake.

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

Schema of the HTTP headers of the opening handshake.

litestar_asyncapi.spec.bindings.WebSocketMessageBinding

Bases: Binding

WebSocket message binding, which defines no protocol-specific fields.

Kafka

litestar_asyncapi.spec.bindings.KafkaServerBinding

Bases: Binding

Kafka server binding naming the schema registry.

schema_registry_url: str | None = None

URL of the schema registry.

schema_registry_vendor: str | None = None

Vendor of the schema registry, such as confluent.

litestar_asyncapi.spec.bindings.KafkaChannelBinding

Bases: Binding

Kafka channel binding describing the topic.

topic: str | None = None

Kafka topic name, when it differs from the channel address.

partitions: int | None = None

Number of partitions of the topic.

replicas: int | None = None

Number of replicas of the topic.

topic_configuration: dict[str, Any] | None = None

Topic configuration properties.

litestar_asyncapi.spec.bindings.KafkaMessageBinding

Bases: Binding

Kafka message binding describing the key and schema id.

key: dict[str, Any] | bool | None = None

Schema of the message key.

schema_id_location: str | None = None

Where the schema id is stored: header or payload.

schema_id_payload_encoding: str | None = None

Number of bytes, or the vendor value, used for a schema id in the payload.

schema_lookup_strategy: str | None = None

Strategy for looking up the schema in the registry.

AMQP

litestar_asyncapi.spec.bindings.AMQPServerBinding

Bases: Binding

AMQP server binding, which defines no protocol-specific fields.

litestar_asyncapi.spec.bindings.AMQPChannelBinding

Bases: Binding

AMQP channel binding describing the queue or exchange behind a channel.

is_: str | None = field(default=None, metadata={'alias': 'is'})

Channel type, serialized as is: queue or routingKey.

queue: dict[str, Any] | None = None

Queue properties when the channel is a queue.

exchange: dict[str, Any] | None = None

Exchange properties when the channel is a routing key.

litestar_asyncapi.spec.bindings.AMQPMessageBinding

Bases: Binding

AMQP message binding.

content_encoding: str | None = None

MIME encoding of the message content.

message_type: str | None = None

Application-specific message type.