Skip to content

RabbitPublisher

faststream.rabbit.publisher.usecase.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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    config: "RabbitPublisherConfig",
    specification: "PublisherSpecification[Any, Any]",
) -> None:
    super().__init__(config, specification)

    self.queue = config.queue
    self.routing_key = config.routing_address.template

    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

queue instance-attribute #

queue = config.queue

routing_key instance-attribute #

routing_key = config.routing_address.template

exchange instance-attribute #

exchange = config.exchange

headers instance-attribute #

headers = config.message_kwargs.pop('headers') or {}

reply_to instance-attribute #

reply_to = config.message_kwargs.pop("reply_to", None) or ""

timeout instance-attribute #

timeout = config.message_kwargs.pop('timeout', None)

publish_options instance-attribute #

publish_options = publish_options

message_options property #

message_options: BasicMessageOptions

is_test instance-attribute #

is_test = False

mock property #

mock: MagicMock

The mock recording the endpoint's calls, available under a test broker.

specification instance-attribute #

specification = specification

routing #

routing(
    *,
    queue: Union[RabbitQueue, str, None] = None,
    routing_key: str = "",
) -> str
Source code in faststream/rabbit/publisher/usecase.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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 
81
82
83
84
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 
 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
114
115
116
117
118
119
120
121
@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", None)
        or self._outer_config.id_generator()
    )

    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 
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
178
179
180
181
182
183
184
@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", None)
        or self._outer_config.id_generator()
    )

    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

assert_called_once_with async #

assert_called_once_with(
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None

Assert the endpoint was called once, with the message described here.

PARAMETER DESCRIPTION
body

The body as a dict, a model or a matcher; it goes through the codec.

TYPE: Any DEFAULT: EMPTY

headers

Headers the message must carry; the rest may carry more.

TYPE: Any DEFAULT: EMPTY

correlation_id

The exact correlation id.

TYPE: Any DEFAULT: EMPTY

reply_to

The exact reply-to destination.

TYPE: Any DEFAULT: EMPTY

content_type

The exact content type.

TYPE: Any DEFAULT: EMPTY

path

The exact path parameters the subject template matched.

TYPE: Any DEFAULT: EMPTY

context

Context paths, as given to Context(), mapped to their values.

TYPE: Mapping[str, Any] DEFAULT: EMPTY

Source code in faststream/_internal/testing/calls.py 
30
31
32
33
34
35
36
37
38
39
40
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
async def assert_called_once_with(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert the endpoint was called once, with the message described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    recorder = self._recorder_with_calls()
    recorder.mock.assert_called_once()
    await recorder.assert_last_call(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

assert_called_with async #

assert_called_with(
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None

Assert the last message the endpoint saw is the one described here.

PARAMETER DESCRIPTION
body

The body as a dict, a model or a matcher; it goes through the codec.

TYPE: Any DEFAULT: EMPTY

headers

Headers the message must carry; the rest may carry more.

TYPE: Any DEFAULT: EMPTY

correlation_id

The exact correlation id.

TYPE: Any DEFAULT: EMPTY

reply_to

The exact reply-to destination.

TYPE: Any DEFAULT: EMPTY

content_type

The exact content type.

TYPE: Any DEFAULT: EMPTY

path

The exact path parameters the subject template matched.

TYPE: Any DEFAULT: EMPTY

context

Context paths, as given to Context(), mapped to their values.

TYPE: Mapping[str, Any] DEFAULT: EMPTY

Source code in faststream/_internal/testing/calls.py 
 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
async def assert_called_with(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert the last message the endpoint saw is the one described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    recorder = self._recorder_with_calls()
    await recorder.assert_last_call(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

assert_any_call async #

assert_any_call(
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None

Assert one of the messages the endpoint saw is the one described here.

PARAMETER DESCRIPTION
body

The body as a dict, a model or a matcher; it goes through the codec.

TYPE: Any DEFAULT: EMPTY

headers

Headers the message must carry; the rest may carry more.

TYPE: Any DEFAULT: EMPTY

correlation_id

The exact correlation id.

TYPE: Any DEFAULT: EMPTY

reply_to

The exact reply-to destination.

TYPE: Any DEFAULT: EMPTY

content_type

The exact content type.

TYPE: Any DEFAULT: EMPTY

path

The exact path parameters the subject template matched.

TYPE: Any DEFAULT: EMPTY

context

Context paths, as given to Context(), mapped to their values.

TYPE: Mapping[str, Any] DEFAULT: EMPTY

Source code in faststream/_internal/testing/calls.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def assert_any_call(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
) -> None:
    """Assert one of the messages the endpoint saw is the one described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
    """
    recorder = self._recorder_with_calls()
    await recorder.assert_any_call(
        ExpectedCall(
            body=body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
        )
    )

set_test #

set_test(
    *, recorder: CallRecorder | None = None, with_fake: bool
) -> None

Turn publisher to testing mode, sharing recorder when one is given.

Source code in faststream/_internal/endpoint/publisher/usecase.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
def set_test(
    self,
    *,
    recorder: CallRecorder | None = None,
    with_fake: bool,
) -> None:
    """Turn publisher to testing mode, sharing `recorder` when one is given."""
    self.is_test = True
    if recorder is None:
        self._recorder.reset()
    else:
        self._recorder = recorder
    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 
67
68
69
70
71
72
73
def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self.is_test = False
    self._recorder.reset()
    # A shared recorder goes back to the handler it belongs to
    self._recorder = CallRecorder(self.specification.name, self._outer_config)
    self._fake_handler = False

schema #

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