Skip to content

RabbitPublisher

faststream.rabbit.publisher.RabbitPublisher #

RabbitPublisher(
    config: RabbitPublisherConfig,
    specification: PublisherSpecification[Any, Any],
)

Bases: PublisherUsecase

A class to represent a RabbitMQ publisher.

Source code in faststream/rabbit/publisher/usecase.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    config: "RabbitPublisherConfig",
    specification: "PublisherSpecification[Any, Any]",
) -> None:
    super().__init__(config, specification)

    self.queue = config.queue
    self.routing_key = config.routing_key

    self.exchange = config.exchange

    self.headers = config.message_kwargs.pop("headers") or {}
    self.reply_to = config.message_kwargs.pop("reply_to", None) or ""
    self.timeout = config.message_kwargs.pop("timeout", None)

    message_options, _ = filter_by_dict(
        BasicMessageOptions,
        dict(config.message_kwargs),
    )
    self._message_options = message_options

    publish_options, _ = filter_by_dict(PublishOptions, dict(config.message_kwargs))
    self.publish_options = publish_options

specification instance-attribute #

specification = specification

middlewares instance-attribute #

middlewares = middlewares

mock instance-attribute #

mock = MagicMock()

queue instance-attribute #

queue = queue

routing_key instance-attribute #

routing_key = routing_key

exchange instance-attribute #

exchange = exchange

headers instance-attribute #

headers = pop('headers') or {}

reply_to instance-attribute #

reply_to = pop('reply_to', None) or ''

timeout instance-attribute #

timeout = pop('timeout', None)

publish_options instance-attribute #

publish_options = publish_options

message_options property #

message_options: BasicMessageOptions

set_test #

set_test(*, mock: MagicMock, with_fake: bool) -> None

Turn publisher to testing mode.

Source code in faststream/_internal/endpoint/publisher/usecase.py 
54
55
56
57
58
59
60
61
62
def set_test(
    self,
    *,
    mock: MagicMock,
    with_fake: bool,
) -> None:
    """Turn publisher to testing mode."""
    self.mock = mock
    self._fake_handler = with_fake

reset_test #

reset_test() -> None

Turn off publisher's testing mode.

Source code in faststream/_internal/endpoint/publisher/usecase.py 
64
65
66
67
def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self._fake_handler = False
    self.mock.reset_mock()

schema #

schema() -> dict[str, PublisherSpec]
Source code in faststream/_internal/endpoint/publisher/usecase.py 
146
147
def schema(self) -> dict[str, "PublisherSpec"]:
    return self.specification.get_schema()

routing #

routing(
    *,
    queue: Union[RabbitQueue, str, None] = None,
    routing_key: str = "",
) -> str
Source code in faststream/rabbit/publisher/usecase.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def routing(
    self,
    *,
    queue: Union["RabbitQueue", str, None] = None,
    routing_key: str = "",
) -> str:
    if not routing_key:
        if q := RabbitQueue.validate(queue):
            routing_key = q.routing()
        else:
            r = self.routing_key or self.queue.routing()
            routing_key = f"{self._outer_config.prefix}{r}"

    return routing_key

start async #

start() -> None
Source code in faststream/rabbit/publisher/usecase.py 
82
83
84
85
async def start(self) -> None:
    if self.exchange is not None:
        await self._outer_config.declarer.declare_exchange(self.exchange)
    return await super().start()

publish async #

publish(
    message: AioPikaSendableMessage,
    queue: Union[RabbitQueue, str, None] = None,
    exchange: Union[RabbitExchange, str, None] = None,
    *,
    routing_key: str = "",
    **publish_kwargs: Unpack[PublishKwargs],
) -> Optional[ConfirmationFrameType]
Source code in faststream/rabbit/publisher/usecase.py 
 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
114
115
116
117
118
119
@override
async def publish(
    self,
    message: "AioPikaSendableMessage",
    queue: Union["RabbitQueue", str, None] = None,
    exchange: Union["RabbitExchange", str, None] = None,
    *,
    routing_key: str = "",
    **publish_kwargs: "Unpack[PublishKwargs]",
) -> Optional["aiormq.abc.ConfirmationFrameType"]:
    if "headers" in publish_kwargs:
        headers = self.headers | (publish_kwargs.pop("headers") or {})
    else:
        headers = self.headers

    correlation_id = publish_kwargs.pop("correlation_id", gen_cor_id())

    cmd = RabbitPublishCommand(
        message,
        routing_key=self.routing(queue=queue, routing_key=routing_key),
        exchange=RabbitExchange.validate(exchange or self.exchange),
        headers=headers,
        correlation_id=correlation_id,
        _publish_type=PublishType.PUBLISH,
        **(self.publish_options | self.message_options | publish_kwargs),  # type: ignore[operator]
    )

    frame: aiormq.abc.ConfirmationFrameType | None = await self._basic_publish(
        cmd,
        producer=self._outer_config.producer,
        _extra_middlewares=(),
    )
    return frame

request async #

request(
    message: AioPikaSendableMessage,
    queue: Union[RabbitQueue, str, None] = None,
    exchange: Union[RabbitExchange, str, None] = None,
    *,
    routing_key: str = "",
    **publish_kwargs: Unpack[PublishKwargs],
) -> RabbitMessage
Source code in faststream/rabbit/publisher/usecase.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@override
async def request(
    self,
    message: "AioPikaSendableMessage",
    queue: Union["RabbitQueue", str, None] = None,
    exchange: Union["RabbitExchange", str, None] = None,
    *,
    routing_key: str = "",
    **publish_kwargs: "Unpack[PublishKwargs]",
) -> "RabbitMessage":
    if "headers" in publish_kwargs:
        headers = self.headers | (publish_kwargs.pop("headers") or {})
    else:
        headers = self.headers

    correlation_id = publish_kwargs.pop("correlation_id", gen_cor_id())

    cmd = RabbitPublishCommand(
        message,
        routing_key=self.routing(queue=queue, routing_key=routing_key),
        exchange=RabbitExchange.validate(exchange or self.exchange),
        correlation_id=correlation_id,
        headers=headers,
        _publish_type=PublishType.PUBLISH,
        **(self.publish_options | self.message_options | publish_kwargs),  # type: ignore[operator]
    )

    msg: RabbitMessage = await self._basic_request(
        cmd,
        producer=self._outer_config.producer,
    )
    return msg