Skip to content

RabbitSubscriberSpecification

faststream.rabbit.subscriber.specification.RabbitSubscriberSpecification #

RabbitSubscriberSpecification(
    _outer_config: T_BrokerConfig,
    specification_config: T_SpecificationConfig,
    calls: CallsCollection[Any],
)

Bases: SubscriberSpecification[RabbitBrokerConfig, RabbitSubscriberSpecificationConfig]

Source code in faststream/_internal/endpoint/subscriber/specification.py 
34
35
36
37
38
39
40
41
def __init__(
    self,
    _outer_config: "T_BrokerConfig",
    specification_config: "T_SpecificationConfig",
    calls: "CallsCollection[Any]",
) -> None:
    super().__init__(_outer_config, specification_config)
    self.calls = calls

channel_labels property #

channel_labels: list[str]

A queue and the exchange it is bound to, which is not the address.

The address is the routing key; a queue is a place to read from. Both are needed to tell two subscribers apart, so the label carries both.

address property #

address: str | None

The routing key the queue is bound by, prefixed.

None where the exchange ignores routing keys: a fanout reaches every queue bound to it, so the queue this subscriber reads from names nothing a message is routed by. The document drops the queue binding for the same reason, and drops the address rather than giving an empty one — AsyncAPI reads an absent address as unknown, which "" does not say.

config instance-attribute #

config = specification_config

include_in_schema property #

include_in_schema: bool

calls instance-attribute #

calls = calls

description property #

description: str | None

call_name property #

call_name: str

name property #

name: str

The key of the first channel this endpoint names.

get_schema #

get_schema() -> dict[str, SubscriberSpec]
Source code in faststream/rabbit/subscriber/specification.py 
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def get_schema(self) -> dict[str, SubscriberSpec]:
    payloads = self.get_payloads()

    queue = self.config.queue.add_prefix(self._outer_config.prefix)

    exchange_binding = amqp.Exchange.from_exchange(self.config.exchange)
    queue_binding = amqp.Queue.from_queue(queue)

    channel_name = self.name

    return {
        channel_name: SubscriberSpec(
            address=self.address,
            description=self.description,
            operation=Operation(
                bindings=OperationBinding(
                    amqp=amqp.OperationBinding(
                        routing_key=queue.routing_template(),
                        queue=queue_binding,
                        exchange=exchange_binding,
                        ack=True,
                        reply_to=None,
                        persist=None,
                        mandatory=None,
                        priority=None,
                    ),
                ),
                message=Message(
                    title=f"{channel_name}:Message",
                    payload=resolve_payloads(payloads),
                ),
            ),
            bindings=ChannelBinding(
                amqp=amqp.ChannelBinding(
                    virtual_host=self._outer_config.virtual_host,
                    queue=queue_binding,
                    exchange=exchange_binding,
                ),
            ),
        ),
    }

get_payloads #

get_payloads() -> list[tuple[dict[str, Any], str]]
Source code in faststream/_internal/endpoint/subscriber/specification.py 
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
def get_payloads(self) -> list[tuple["dict[str, Any]", str]]:
    payloads: list[tuple[dict[str, Any], str]] = []

    call_name = self.call_name

    for h in self.calls:
        if h.dependant is None:
            msg = "You should setup `Handler` at first."
            raise SetupError(msg)

        body = parse_handler_params(
            h.dependant,
            prefix=f"{self.config.title_ or call_name}:Message",
        )
        payloads.append((body, to_camelcase(h.name)))

    if not self.calls:
        payloads.append(
            (
                {
                    "title": f"{self.config.title_ or call_name}:Message:Payload",
                },
                to_camelcase(call_name),
            ),
        )

    return payloads