Skip to content

OperationBinding

faststream.specification.asyncapi.v3_0_0.schema.bindings.main.operation.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: OperationBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v3_0_0/schema/bindings/main/operation.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@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: OperationBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v3_0_0/schema/bindings/main/operation.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@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