Skip to content

Channel

faststream.specification.asyncapi.v3_0_0.schema.channels.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 class-attribute instance-attribute #

address: str | None = None

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(
    subscriber: SubscriberSpec,
    servers: list[dict[str, str]] | None = None,
) -> Self
Source code in faststream/specification/asyncapi/v3_0_0/schema/channels.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@classmethod
def from_sub(
    cls,
    subscriber: SubscriberSpec,
    servers: list[dict[str, str]] | None = None,
) -> 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=subscriber.address,
        messages={
            "SubscribeMessage": Message.from_spec(message),
        },
        bindings=ChannelBinding.from_sub(subscriber.bindings),
        servers=servers,
    )

from_pub classmethod #

from_pub(
    publisher: PublisherSpec,
    servers: list[dict[str, str]] | None = None,
) -> Self
Source code in faststream/specification/asyncapi/v3_0_0/schema/channels.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@classmethod
def from_pub(
    cls,
    publisher: PublisherSpec,
    servers: list[dict[str, str]] | None = None,
) -> Self:
    return cls(
        description=publisher.description,
        address=publisher.address,
        messages={
            "Message": Message.from_spec(publisher.operation.message),
        },
        bindings=ChannelBinding.from_pub(publisher.bindings),
        servers=servers,
    )