Skip to content

OperationBinding

faststream.specification.asyncapi.v3_0_0.schema.bindings.amqp.OperationBinding #

Bases: BaseModel

cc class-attribute instance-attribute #

cc: list[str] | None = None

ack instance-attribute #

ack: bool

replyTo class-attribute instance-attribute #

replyTo: str | None = None

deliveryMode class-attribute instance-attribute #

deliveryMode: int | None = None

mandatory class-attribute instance-attribute #

mandatory: bool | None = None

priority class-attribute instance-attribute #

priority: PositiveInt | None = None

bindingVersion class-attribute instance-attribute #

bindingVersion: str = '0.3.0'

from_sub classmethod #

from_sub(binding: OperationBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v3_0_0/schema/bindings/amqp/operation.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@classmethod
def from_sub(cls, binding: amqp.OperationBinding | None) -> Self | None:
    if not binding:
        return None

    return cls(
        cc=[binding.routing_key]
        if (binding.routing_key and binding.exchange.is_respect_routing_key)
        else None,
        ack=binding.ack,
        replyTo=binding.reply_to,
        deliveryMode=None if binding.persist is None else int(binding.persist) + 1,
        mandatory=binding.mandatory,
        priority=binding.priority,
    )

from_pub classmethod #

from_pub(binding: OperationBinding | None) -> Self | None
Source code in faststream/specification/asyncapi/v3_0_0/schema/bindings/amqp/operation.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@classmethod
def from_pub(cls, binding: amqp.OperationBinding | None) -> Self | None:
    if not binding:
        return None

    return cls(
        cc=None
        if (not binding.routing_key or not binding.exchange.is_respect_routing_key)
        else [binding.routing_key],
        ack=binding.ack,
        replyTo=binding.reply_to,
        deliveryMode=None if binding.persist is None else int(binding.persist) + 1,
        mandatory=binding.mandatory,
        priority=binding.priority,
    )