Skip to content

Channel

faststream.specification.asyncapi.v3_0_0.schema.Channel #

Bases: BaseModel

A class to represent a channel.

Configurations

model_config : configuration for the model (only applicable for Pydantic version 2) Config : configuration for the class (only applicable for Pydantic version 1)

address instance-attribute #

address: str

description class-attribute instance-attribute #

description: str | None = None

servers class-attribute instance-attribute #

servers: list[dict[str, str]] | None = None

messages instance-attribute #

messages: dict[str, Message | Reference]

bindings class-attribute instance-attribute #

bindings: 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(address: str, subscriber: SubscriberSpec) -> Self
Source code in faststream/specification/asyncapi/v3_0_0/schema/channels.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@classmethod
def from_sub(cls, address: str, subscriber: SubscriberSpec) -> Self:
    message = subscriber.operation.message
    assert message.title

    *left, right = message.title.split(":")
    message.title = ":".join((*left, f"Subscribe{right}"))

    return cls(
        description=subscriber.description,
        address=address,
        messages={
            "SubscribeMessage": Message.from_spec(message),
        },
        bindings=ChannelBinding.from_sub(subscriber.bindings),
        servers=None,
    )

from_pub classmethod #

from_pub(address: str, publisher: PublisherSpec) -> Self
Source code in faststream/specification/asyncapi/v3_0_0/schema/channels.py 
62
63
64
65
66
67
68
69
70
71
72
@classmethod
def from_pub(cls, address: str, publisher: PublisherSpec) -> Self:
    return cls(
        description=publisher.description,
        address=address,
        messages={
            "Message": Message.from_spec(publisher.operation.message),
        },
        bindings=ChannelBinding.from_pub(publisher.bindings),
        servers=None,
    )