Skip to content

Channel

faststream.specification.asyncapi.v2_6_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)

description class-attribute instance-attribute #

description: str | None = None

servers class-attribute instance-attribute #

servers: list[str] | None = None

bindings class-attribute instance-attribute #

bindings: ChannelBinding | None = None

subscribe class-attribute instance-attribute #

subscribe: Operation | None = None

publish class-attribute instance-attribute #

publish: Operation | 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[str] | None = None,
) -> Self
Source code in faststream/specification/asyncapi/v2_6_0/schema/channels.py 
43
44
45
46
47
48
49
50
51
52
53
@classmethod
def from_sub(
    cls, subscriber: SubscriberSpec, servers: list[str] | None = None
) -> Self:
    return cls(
        description=subscriber.description,
        servers=servers,
        bindings=ChannelBinding.from_sub(subscriber.bindings),
        subscribe=None,
        publish=Operation.from_sub(subscriber.operation),
    )

from_pub classmethod #

from_pub(
    publisher: PublisherSpec,
    servers: list[str] | None = None,
) -> Self
Source code in faststream/specification/asyncapi/v2_6_0/schema/channels.py 
55
56
57
58
59
60
61
62
63
@classmethod
def from_pub(cls, publisher: PublisherSpec, servers: list[str] | None = None) -> Self:
    return cls(
        description=publisher.description,
        servers=servers,
        bindings=ChannelBinding.from_pub(publisher.bindings),
        subscribe=Operation.from_pub(publisher.operation),
        publish=None,
    )