Skip to content

RabbitPublisherSpecification

faststream.rabbit.publisher.specification.RabbitPublisherSpecification #

RabbitPublisherSpecification(
    _outer_config: T_BrokerConfig,
    specification_config: T_SpecificationConfig,
)

Bases: PublisherSpecification[RabbitBrokerConfig, RabbitPublisherSpecificationConfig]

Source code in faststream/_internal/endpoint/publisher/specification.py 
29
30
31
32
33
34
35
36
def __init__(
    self,
    _outer_config: "T_BrokerConfig",
    specification_config: "T_SpecificationConfig",
) -> None:
    super().__init__(_outer_config, specification_config)

    self.calls: list[AnyCallable] = []

routing property #

routing: str | None

The routing key this publisher was declared with, as the name reads it.

This is the channel name's question, not the address's: an explicit routing_key names the channel whatever the exchange does with it. See address for the string a message actually travels by.

address property #

address: str | None

The routing key a message published here travels by, prefixed.

The exchange decides first, not the declaration: a fanout reaches every queue bound to it and ignores any routing key handed to it, so one declared anyway still addresses nothing. The subscriber gates on the same question, which is what keeps both ends of an address showing one string.

None rather than "" for that case: AsyncAPI reads an absent address as unknown, and an empty one as an address zero characters long.

name property #

name: str

config instance-attribute #

config = specification_config

include_in_schema property #

include_in_schema: bool

calls instance-attribute #

calls: list[AnyCallable] = []

get_schema #

get_schema() -> dict[str, PublisherSpec]
Source code in faststream/rabbit/publisher/specification.py 
 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
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def get_schema(self) -> dict[str, "PublisherSpec"]:
    payloads = self.get_payloads()

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

    # deliberately not `self.address`: the binding hands the routing key over
    # whatever the exchange type, and the renderer is what drops it where the
    # exchange ignores one. `address` answers for the document instead, so it
    # has to make that call itself.
    r = self.config.routing_address.template or self.config.queue.routing_template()
    routing_key = f"{self._outer_config.prefix}{r}"

    return {
        self.name: PublisherSpec(
            address=self.address,
            description=self.config.description_,
            operation=Operation(
                bindings=OperationBinding(
                    amqp=amqp.OperationBinding(
                        routing_key=routing_key or None,
                        queue=queue_binding,
                        exchange=exchange_binding,
                        ack=True,
                        persist=self.config.message_kwargs.get("persist"),
                        priority=self.config.message_kwargs.get("priority"),
                        reply_to=self.config.message_kwargs.get("reply_to"),
                        mandatory=self.config.message_kwargs.get("mandatory"),
                    ),
                ),
                message=Message(
                    title=f"{self.name}:Message",
                    payload=resolve_payloads(
                        payloads,
                        "Publisher",
                        served_words=2 if self.config.title_ is None else 1,
                    ),
                ),
            ),
            bindings=ChannelBinding(
                amqp=amqp.ChannelBinding(
                    virtual_host=self._outer_config.virtual_host,
                    queue=queue_binding,
                    exchange=exchange_binding,
                ),
            ),
        ),
    }

add_call #

add_call(call: AnyCallable) -> None
Source code in faststream/_internal/endpoint/publisher/specification.py 
38
39
def add_call(self, call: "AnyCallable") -> None:
    self.calls.append(call)

get_payloads #

get_payloads() -> list[tuple[dict[str, Any], str]]
Source code in faststream/_internal/endpoint/publisher/specification.py 
41
42
43
44
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
79
80
81
82
83
84
85
86
87
def get_payloads(self) -> list[tuple[dict[str, Any], str]]:
    payloads: list[tuple[dict[str, Any], str]] = []

    if self.config.schema_:
        body = get_model_schema(
            call=create_model(
                "",
                __config__=get_config_base(),
                response__=(self.config.schema_, ...),
            ),
            prefix=f"{self.name}:Message",
        )

        if body:  # pragma: no branch
            payloads.append((body, ""))

    else:
        di_state = self._outer_config.fd_config

        for call in self.calls:
            call_model = build_call_model(
                call,
                dependency_provider=di_state.provider,
                serializer_cls=di_state._serializer,
            )

            if call_model.serializer:
                response_type = next(
                    iter(call_model.serializer.response_option.values()),
                ).field_type
            else:
                response_type = None

            if response_type is not None and response_type is not Parameter.empty:
                body = get_model_schema(
                    create_model(
                        "",
                        __config__=get_config_base(),
                        response__=(response_type, ...),
                    ),
                    prefix=f"{self.name}:Message",
                )

                if body:
                    payloads.append((body, to_camelcase(unwrap(call).__name__)))

    return payloads