RabbitQueue

def __init__(
    self,
    name: str,
    queue_type: QueueType = QueueType.CLASSIC,
    durable: bool = EMPTY,
    exclusive: bool = False,
    declare: bool = True,
    passive: Annotated[
        bool,
        deprecated("Use `declare` instead. Will be removed in the 0.6.0 release."),
        Doc("Do not create queue automatically."),
    ] = EMPTY,
    auto_delete: bool = False,
    arguments: Union[
        "QuorumQueueArgs",
        "ClassicQueueArgs",
        "StreamQueueArgs",
        "AnyDict",
        None,
    ] = None,
    timeout: "TimeoutType" = None,
    robust: bool = True,
    bind_arguments: Optional["AnyDict"] = 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 passive: Do not create queue automatically.
    :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:
            raise SetupError("Quorum and Stream queues must be durable")
    elif durable is EMPTY:
        durable = False

    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.passive = passive
    self.auto_delete = auto_delete
    self.arguments = {"x-queue-type": queue_type.value, **(arguments or {})}
    self.timeout = timeout

    if passive is not EMPTY:
        warnings.warn(
            DeprecationWarning(
                "Use `declare` instead. Will be removed in the 0.6.0 release.",
            ),
            stacklevel=2,
        )
        self.declare = not passive
    else:
        self.declare = declare