Skip to content

RabbitQueue

faststream.rabbit.schemas.RabbitQueue #

RabbitQueue(
    name: str,
    queue_type: Literal[CLASSIC] = CLASSIC,
    durable: bool = EMPTY,
    exclusive: bool = False,
    declare: bool = True,
    auto_delete: bool = False,
    arguments: Optional[ClassicQueueArgs] = None,
    timeout: TimeoutType = None,
    robust: bool = True,
    bind_arguments: dict[str, Any] | None = None,
    routing_key: str = "",
)

RabbitQueue(
    name: str,
    queue_type: Literal[QUORUM],
    durable: Literal[True] = EMPTY,
    exclusive: bool = False,
    declare: bool = True,
    auto_delete: bool = False,
    arguments: Optional[QuorumQueueArgs] = None,
    timeout: TimeoutType = None,
    robust: bool = True,
    bind_arguments: dict[str, Any] | None = None,
    routing_key: str = "",
)

RabbitQueue(
    name: str,
    queue_type: Literal[STREAM],
    durable: Literal[True] = EMPTY,
    exclusive: bool = False,
    declare: bool = True,
    auto_delete: bool = False,
    arguments: Optional[StreamQueueArgs] = None,
    timeout: TimeoutType = None,
    robust: bool = True,
    bind_arguments: dict[str, Any] | None = None,
    routing_key: str = "",
)

RabbitQueue(
    name: str,
    queue_type: QueueType = CLASSIC,
    durable: bool = EMPTY,
    exclusive: bool = False,
    declare: bool = True,
    auto_delete: bool = False,
    arguments: Union[
        QuorumQueueArgs,
        ClassicQueueArgs,
        StreamQueueArgs,
        dict[str, Any],
        None,
    ] = None,
    timeout: TimeoutType = None,
    robust: bool = True,
    bind_arguments: dict[str, Any] | None = None,
    routing_key: str = "",
)

Bases: NameRequired

A class to represent a RabbitMQ queue.

You can find information about all options in the official RabbitMQ documentation:

https://www.rabbitmq.com/docs/queues

Initialize the RabbitMQ queue.

:param name: RabbitMQ queue name. :param durable: Whether the object is durable. :param exclusive: The queue can be used only in the current connection and will be deleted after connection closed. :param declare: Whether to queue automatically or just connect to it. If you want to connect to an existing queue, set this to False. Copy of passive aio-pike option. :param auto_delete: The queue will be deleted after connection closed. :param arguments: Queue declaration arguments. You can find information about them in the official RabbitMQ documentation: https://www.rabbitmq.com/docs/queues#optional-arguments :param timeout: Send confirmation time from RabbitMQ. :param robust: Whether to declare queue object as restorable. :param bind_arguments: Queue-exchange binding options. :param routing_key: Explicit binding routing key. Uses name if not present.

Source code in faststream/rabbit/schemas/queue.py 
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
225
226
227
def __init__(
    self,
    name: str,
    queue_type: QueueType = QueueType.CLASSIC,
    durable: bool = EMPTY,
    exclusive: bool = False,
    declare: bool = True,
    auto_delete: bool = False,
    arguments: Union[
        "QuorumQueueArgs",
        "ClassicQueueArgs",
        "StreamQueueArgs",
        dict[str, Any],
        None,
    ] = None,
    timeout: "TimeoutType" = None,
    robust: bool = True,
    bind_arguments: dict[str, Any] | None = None,
    routing_key: str = "",
) -> None:
    """Initialize the RabbitMQ queue.

    :param name: RabbitMQ queue name.
    :param durable: Whether the object is durable.
    :param exclusive: The queue can be used only in the current connection and will be deleted after connection closed.
    :param declare: Whether to queue automatically or just connect to it.
                    If you want to connect to an existing queue, set this to `False`.
                    Copy of `passive` aio-pike option.
    :param auto_delete: The queue will be deleted after connection closed.
    :param arguments: Queue declaration arguments.
                      You can find information about them in the official RabbitMQ documentation:
                      https://www.rabbitmq.com/docs/queues#optional-arguments
    :param timeout: Send confirmation time from RabbitMQ.
    :param robust: Whether to declare queue object as restorable.
    :param bind_arguments: Queue-exchange binding options.
    :param routing_key: Explicit binding routing key. Uses name if not present.
    """
    if queue_type is QueueType.QUORUM or queue_type is QueueType.STREAM:
        if durable is EMPTY:
            durable = True
        elif not durable:
            error_msg = "Quorum and Stream queues must be durable"
            raise SetupError(error_msg)
    elif durable is EMPTY:
        durable = True

    super().__init__(name)

    self.durable = durable
    self.exclusive = exclusive
    self.bind_arguments = bind_arguments
    self.routing_address = Address(routing_key, RABBIT_ADDRESS_SYNTAX)
    self.robust = robust
    self.auto_delete = auto_delete
    self.arguments = {"x-queue-type": queue_type.value, **(arguments or {})}
    self.timeout = timeout
    self.declare = declare

name instance-attribute #

name = name

routing_key property #

routing_key: str

The routing key as it reaches RabbitMQ.

path_regex property #

path_regex: Optional[Pattern[str]]

durable instance-attribute #

durable = durable

exclusive instance-attribute #

exclusive = exclusive

bind_arguments instance-attribute #

bind_arguments = bind_arguments

routing_address instance-attribute #

routing_address = Address(
    routing_key, RABBIT_ADDRESS_SYNTAX
)

robust instance-attribute #

robust = robust

auto_delete instance-attribute #

auto_delete = auto_delete

arguments instance-attribute #

arguments = {'x-queue-type': queue_type.value, **arguments or {}}

timeout instance-attribute #

timeout = timeout

declare instance-attribute #

declare = declare

validate classmethod #

validate(value: str | Self, **kwargs: Any) -> Self

validate(value: None, **kwargs: Any) -> None

validate(
    value: str | Self | None, **kwargs: Any
) -> Self | None

Factory to create object.

Source code in faststream/_internal/proto.py 
30
31
32
33
34
35
@classmethod
def validate(cls, value: str | Self | None, **kwargs: Any) -> Self | None:
    """Factory to create object."""
    if value is not None and isinstance(value, str):
        value = cls(value, **kwargs)
    return value

routing #

routing() -> str

Return the Broker address of object.

Source code in faststream/rabbit/schemas/queue.py 
103
104
105
def routing(self) -> str:
    """Return the Broker address of object."""
    return self.routing_address.broker_address or self.name

routing_template #

routing_template() -> str

Return the Address template of object.

Source code in faststream/rabbit/schemas/queue.py 
107
108
109
110
111
def routing_template(self) -> str:
    """Return the Address template of object."""
    if self.routing_address:
        return self.routing_address.template
    return self.name

add_prefix #

add_prefix(prefix: str) -> RabbitQueue
Source code in faststream/rabbit/schemas/queue.py 
113
114
115
116
117
118
119
120
121
def add_prefix(self, prefix: str) -> "RabbitQueue":
    new_q: RabbitQueue = deepcopy(self)

    new_q.name = f"{prefix}{new_q.name}"

    if new_q.routing_address:
        new_q.routing_address = new_q.routing_address.add_prefix(prefix)

    return new_q