RabbitQueue

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.
    """
    re, routing_key = compile_path(
        routing_key,
        replace_symbol="*",
        patch_regex=lambda x: x.replace(r"\#", ".+"),
    )

    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.path_regex = re
    self.durable = durable
    self.exclusive = exclusive
    self.bind_arguments = bind_arguments
    self.routing_key = routing_key
    self.robust = robust
    self.auto_delete = auto_delete
    self.arguments = {"x-queue-type": queue_type.value, **(arguments or {})}
    self.timeout = timeout
    self.declare = declare