Skip to content

ChannelBinding

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

Bases: BaseModel

A class to represent channel binding.

is_ class-attribute instance-attribute #

is_: Literal["queue", "routingKey"] = Field(..., alias="is")

bindingVersion class-attribute instance-attribute #

bindingVersion: str = '0.2.0'

queue class-attribute instance-attribute #

queue: Queue | None = None

exchange class-attribute instance-attribute #

exchange: Exchange | None = None

from_sub classmethod #

from_sub(binding: ChannelBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v2_6_0/schema/bindings/amqp/channel.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@classmethod
def from_sub(cls, binding: amqp.ChannelBinding | None) -> Self | None:
    if binding is None:
        return None

    return cls(
        **{
            "is": "routingKey",
            "queue": Queue.from_spec(binding.queue, binding.virtual_host)
            if binding.exchange.is_respect_routing_key
            else None,
            "exchange": Exchange.from_spec(binding.exchange, binding.virtual_host),
        },
    )

from_pub classmethod #

from_pub(binding: ChannelBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v2_6_0/schema/bindings/amqp/channel.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@classmethod
def from_pub(cls, binding: amqp.ChannelBinding | None) -> Self | None:
    if binding is None:
        return None

    return cls(
        **{
            "is": "routingKey",
            "queue": Queue.from_spec(binding.queue, binding.virtual_host)
            if binding.exchange.is_respect_routing_key and binding.queue.name
            else None,
            "exchange": Exchange.from_spec(binding.exchange, binding.virtual_host),
        },
    )