Skip to content

OperationBinding

faststream.specification.asyncapi.v2_6_0.schema.bindings.mqtt.OperationBinding #

Bases: BaseModel

MQTT operation binding (AsyncAPI 2.6.0).

qos class-attribute instance-attribute #

qos: int = 0

retain class-attribute instance-attribute #

retain: bool = False

messageExpiryInterval class-attribute instance-attribute #

messageExpiryInterval: int | None = None

bindingVersion class-attribute instance-attribute #

bindingVersion: str = '0.2.0'

from_sub classmethod #

from_sub(binding: OperationBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v2_6_0/schema/bindings/mqtt/operation.py 
27
28
29
30
31
32
33
34
35
36
37
@classmethod
def from_sub(cls, binding: mqtt.OperationBinding | None) -> Self | None:
    if binding is None:
        return None

    return cls(
        qos=binding.qos,
        retain=binding.retain,
        messageExpiryInterval=binding.messageExpiryInterval,
        bindingVersion=binding.bindingVersion,
    )

from_pub classmethod #

from_pub(binding: OperationBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v2_6_0/schema/bindings/mqtt/operation.py 
39
40
41
42
43
44
45
46
47
48
49
@classmethod
def from_pub(cls, binding: mqtt.OperationBinding | None) -> Self | None:
    if binding is None:
        return None

    return cls(
        qos=binding.qos,
        retain=binding.retain,
        messageExpiryInterval=binding.messageExpiryInterval,
        bindingVersion=binding.bindingVersion,
    )