Skip to content

OperationBinding

faststream.specification.asyncapi.v2_6_0.schema.bindings.main.OperationBinding #

Bases: BaseModel

A class to represent an operation binding.

amqp class-attribute instance-attribute #

amqp: OperationBinding | None = None

kafka class-attribute instance-attribute #

kafka: OperationBinding | None = None

mqtt class-attribute instance-attribute #

mqtt: OperationBinding | None = None

sqs class-attribute instance-attribute #

sqs: OperationBinding | None = None

nats class-attribute instance-attribute #

nats: OperationBinding | None = None

redis class-attribute instance-attribute #

redis: OperationBinding | None = None

http class-attribute instance-attribute #

http: OperationBinding | 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(binding: None) -> None

from_sub(binding: OperationBinding) -> Self

from_sub(binding: OperationBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v2_6_0/schema/bindings/main/operation.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@classmethod
def from_sub(cls, binding: SpecBinding | None) -> Self | None:
    if binding is None:
        return None

    if binding.amqp and (
        amqp := amqp_bindings.OperationBinding.from_sub(binding.amqp)
    ):
        return cls(amqp=amqp)

    if binding.kafka and (
        kafka := kafka_bindings.OperationBinding.from_sub(binding.kafka)
    ):
        return cls(kafka=kafka)

    if binding.mqtt and (
        mqtt := mqtt_bindings.OperationBinding.from_sub(binding.mqtt)
    ):
        return cls(mqtt=mqtt)

    if binding.nats and (
        nats := nats_bindings.OperationBinding.from_sub(binding.nats)
    ):
        return cls(nats=nats)

    if binding.redis and (
        redis := redis_bindings.OperationBinding.from_sub(binding.redis)
    ):
        return cls(redis=redis)

    if binding.sqs:
        return cls(sqs=sqs_bindings.OperationBinding.from_sub(binding.sqs))

    return None

from_pub classmethod #

from_pub(binding: None) -> None

from_pub(binding: OperationBinding) -> Self

from_pub(binding: OperationBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v2_6_0/schema/bindings/main/operation.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@classmethod
def from_pub(cls, binding: SpecBinding | None) -> Self | None:
    if binding is None:
        return None

    if binding.amqp and (
        amqp := amqp_bindings.OperationBinding.from_pub(binding.amqp)
    ):
        return cls(amqp=amqp)

    if binding.kafka and (
        kafka := kafka_bindings.OperationBinding.from_pub(binding.kafka)
    ):
        return cls(kafka=kafka)

    if binding.mqtt and (
        mqtt := mqtt_bindings.OperationBinding.from_pub(binding.mqtt)
    ):
        return cls(mqtt=mqtt)

    if binding.nats and (
        nats := nats_bindings.OperationBinding.from_pub(binding.nats)
    ):
        return cls(nats=nats)

    if binding.redis and (
        redis := redis_bindings.OperationBinding.from_pub(binding.redis)
    ):
        return cls(redis=redis)

    if binding.sqs:
        return cls(sqs=sqs_bindings.OperationBinding.from_pub(binding.sqs))

    return None