Skip to content

ChannelBinding

faststream.specification.asyncapi.v3_0_0.schema.bindings.ChannelBinding #

Bases: BaseModel

A class to represent channel bindings.

ATTRIBUTE DESCRIPTION
amqp

AMQP channel binding (optional)
kafka

Kafka channel binding (optional)
sqs

SQS channel binding (optional)
nats

NATS channel binding (optional)
redis

Redis channel binding (optional)

amqp class-attribute instance-attribute #

amqp = None

kafka class-attribute instance-attribute #

kafka = None

sqs class-attribute instance-attribute #

sqs = None

nats class-attribute instance-attribute #

nats = None

redis class-attribute instance-attribute #

redis = 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)
Source code in faststream/specification/asyncapi/v3_0_0/schema/bindings/main/channel.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@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.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 and (sqs := sqs_bindings.ChannelBinding.from_sub(binding.sqs)):
        return cls(sqs=sqs)

    return None

from_pub classmethod #

from_pub(binding)
Source code in faststream/specification/asyncapi/v3_0_0/schema/bindings/main/channel.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@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.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 and (sqs := sqs_bindings.ChannelBinding.from_pub(binding.sqs)):
        return cls(sqs=sqs)

    return None