Skip to content

RabbitRoute

faststream.rabbit.RabbitRoute #

RabbitRoute(
    call: Callable[..., AioPikaSendableMessage]
    | Callable[..., Awaitable[AioPikaSendableMessage]],
    queue: Union[str, RabbitQueue],
    exchange: Union[str, RabbitExchange, None] = None,
    *,
    publishers: Iterable[RabbitPublisher] = (),
    consume_args: dict[str, Any] | None = None,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    persistent: bool = True,
    codec: Optional[CodecProto] = None,
    channel: Optional[Channel] = None,
)

Bases: SubscriberRoute

Class to store delayed RabbitBroker subscriber registration.

Just a copy of RabbitRegistrator.subscriber(...) arguments.

Initialized RabbitRoute.

PARAMETER DESCRIPTION
call

Message handler function to wrap the same with @broker.subscriber(...) way.

TYPE: Callable[..., AioPikaSendableMessage] | Callable[..., Awaitable[AioPikaSendableMessage]]

queue

RabbitMQ queue to listen. FastStream declares and binds queue object to exchange automatically by default.

TYPE: Union[str, RabbitQueue]

exchange

RabbitMQ exchange to bind queue to. Uses default exchange if not present. FastStream declares exchange object automatically by default.

TYPE: Union[str, RabbitExchange, None] DEFAULT: None

publishers

RabbitMQ publishers to broadcast the handler result.

TYPE: Iterable[RabbitPublisher] DEFAULT: ()

consume_args

Extra consumer arguments to use in queue.consume(...) method.

TYPE: dict[str, Any] | None DEFAULT: None

dependencies

Dependencies list ([Dependant(),]) to apply to the subscriber.

TYPE: Sequence[Dependant] DEFAULT: ()

parser

Parser to map original IncomingMessage Msg to FastStream one.

TYPE: Optional[CustomCallable] DEFAULT: None

decoder

Function to decode FastStream msg bytes body to python objects.

TYPE: Optional[CustomCallable] DEFAULT: None

ack_policy

Acknowledgment policy for the subscriber (by default MANUAL).

TYPE: AckPolicy DEFAULT: EMPTY

no_reply

Whether to disable FastStream RPC and Reply To auto responses or not.

TYPE: bool DEFAULT: False

title

AsyncAPI subscriber object title.

TYPE: str | None DEFAULT: None

description

AsyncAPI subscriber object description. Uses decorated docstring as default.

TYPE: str | None DEFAULT: None

include_in_schema

Whetever to include operation in AsyncAPI schema or not.

TYPE: bool DEFAULT: True

persistent

Whether to make the subscriber persistent or not.

TYPE: bool DEFAULT: True

codec

Custom codec object.

TYPE: Optional[CodecProto] DEFAULT: None

channel

Channel to use for consuming messages.

TYPE: Optional[Channel] DEFAULT: None

Source code in faststream/rabbit/broker/router.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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def __init__(
    self,
    call: Callable[..., "AioPikaSendableMessage"]
    | Callable[..., Awaitable["AioPikaSendableMessage"]],
    queue: Union[str, "RabbitQueue"],
    exchange: Union[str, "RabbitExchange", None] = None,
    *,
    publishers: Iterable[RabbitPublisher] = (),
    consume_args: dict[str, Any] | None = None,
    # broker arguments
    dependencies: Sequence["Dependant"] = (),
    parser: Optional["CustomCallable"] = None,
    decoder: Optional["CustomCallable"] = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    # AsyncAPI information
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    persistent: bool = True,
    codec: Optional["CodecProto"] = None,
    channel: Optional["Channel"] = None,
) -> None:
    """Initialized RabbitRoute.

    Args:
        call:
            Message handler function
            to wrap the same with `@broker.subscriber(...)` way.
        queue:
            RabbitMQ queue to listen.
            **FastStream** declares and binds queue object to `exchange` automatically by default.
        exchange:
            RabbitMQ exchange to bind queue to.
            Uses default exchange if not present.
            **FastStream** declares exchange object automatically by default.
        publishers:
            RabbitMQ publishers to broadcast the handler result.
        consume_args:
            Extra consumer arguments to use in `queue.consume(...)` method.
        dependencies:
            Dependencies list (`[Dependant(),]`) to apply to the subscriber.
        parser:
            Parser to map original **IncomingMessage** Msg to FastStream one.
        decoder:
            Function to decode FastStream msg bytes body to python objects.
        ack_policy:
            Acknowledgment policy for the subscriber (by default `MANUAL`).
        no_reply:
            Whether to disable **FastStream** RPC and Reply To auto responses or not.
        title:
            AsyncAPI subscriber object title.
        description:
            AsyncAPI subscriber object description.
            Uses decorated docstring as default.
        include_in_schema:
            Whetever to include operation in AsyncAPI schema or not.
        persistent: Whether to make the subscriber persistent or not.
        codec: Custom codec object.
        channel: Channel to use for consuming messages.
    """
    super().__init__(
        call,
        publishers=publishers,
        queue=queue,
        exchange=exchange,
        consume_args=consume_args,
        dependencies=dependencies,
        parser=parser,
        decoder=decoder,
        ack_policy=ack_policy,
        no_reply=no_reply,
        title=title,
        description=description,
        include_in_schema=include_in_schema,
        persistent=persistent,
        codec=codec,
        channel=channel,
    )

args instance-attribute #

args: Iterable[Any] = args

kwargs instance-attribute #

kwargs: dict[str, Any] = kwargs

call instance-attribute #

call: Callable[..., Any] = call

publishers instance-attribute #

publishers: Iterable[Any] = publishers