Skip to content

ChannelBinding

faststream.specification.asyncapi.v2_6_0.schema.bindings.main.channel.ChannelBinding #

Bases: BaseModel

A class to represent channel bindings.

amqp class-attribute instance-attribute #

amqp: ChannelBinding | None = None

kafka class-attribute instance-attribute #

kafka: ChannelBinding | None = None

mqtt class-attribute instance-attribute #

mqtt: ChannelBinding | None = None

sqs class-attribute instance-attribute #

sqs: ChannelBinding | None = None

nats class-attribute instance-attribute #

nats: ChannelBinding | None = None

redis class-attribute instance-attribute #

redis: ChannelBinding | None = None

model_config class-attribute instance-attribute #

model_config = {'extra': 'allow'}

Config #

extra class-attribute instance-attribute #

extra = 'allow'

from_sub classmethod #

from_sub(binding: None) -> None

from_sub(binding: ChannelBinding) -> Self

from_sub(binding: ChannelBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v2_6_0/schema/bindings/main/channel.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@classmethod
def from_sub(cls, binding: SpecBinding | None) -> Self | None:
    if binding is None:
        return None

    if binding.amqp and (amqp := amqp_bindings.ChannelBinding.from_sub(binding.amqp)):
        return cls(amqp=amqp)

    if binding.kafka and (
        kafka := kafka_bindings.ChannelBinding.from_sub(binding.kafka)
    ):
        return cls(kafka=kafka)

    if binding.mqtt and (mqtt := mqtt_bindings.ChannelBinding.from_sub(binding.mqtt)):
        return cls(mqtt=mqtt)

    if binding.nats and (nats := nats_bindings.ChannelBinding.from_sub(binding.nats)):
        return cls(nats=nats)

    if binding.redis and (
        redis := redis_bindings.ChannelBinding.from_sub(binding.redis)
    ):
        return cls(redis=redis)

    if binding.sqs:
        return cls(sqs=sqs_bindings.ChannelBinding.from_sub(binding.sqs))

    return None

from_pub classmethod #

from_pub(binding: None) -> None

from_pub(binding: ChannelBinding) -> Self

from_pub(binding: ChannelBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v2_6_0/schema/bindings/main/channel.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@classmethod
def from_pub(cls, binding: SpecBinding | None) -> Self | None:
    if binding is None:
        return None

    if binding.amqp and (amqp := amqp_bindings.ChannelBinding.from_pub(binding.amqp)):
        return cls(amqp=amqp)

    if binding.kafka and (
        kafka := kafka_bindings.ChannelBinding.from_pub(binding.kafka)
    ):
        return cls(kafka=kafka)

    if binding.mqtt and (mqtt := mqtt_bindings.ChannelBinding.from_pub(binding.mqtt)):
        return cls(mqtt=mqtt)

    if binding.nats and (nats := nats_bindings.ChannelBinding.from_pub(binding.nats)):
        return cls(nats=nats)

    if binding.redis and (
        redis := redis_bindings.ChannelBinding.from_pub(binding.redis)
    ):
        return cls(redis=redis)

    if binding.sqs:
        return cls(sqs=sqs_bindings.ChannelBinding.from_pub(binding.sqs))

    return None