Exchanges and Queues

Exchanges, queues and bindings describe the broker topology a producer publishes into and a consumer reads from.

Exchange

Exchange

An Exchange declaration.

Arguments: name: Name of the exchange. Default is '' (default exchange). type: Exchange type ('direct', 'fanout', 'topic'). Default is 'direct'. durable: Survive broker restart. Default is True. auto_delete: Delete when no queues bound. Default is False. delivery_mode: Default delivery mode for messages. arguments: Additional exchange arguments. no_declare: Never declare this exchange.

In Redis, exchanges are emulated: - direct: Messages go to queue matching routing_key - fanout: Messages go to all bound queues (via pub/sub) - topic: Messages match queue patterns

can_cache_declaration property

can_cache_declaration: bool

Whether one declaration holds for as long as the channel does.

The broker drops an auto-delete exchange once its last queue lets go of it, so that one has to be declared every time, as upstream does.

declare async

declare(channel: Channel | None = None) -> None

Declare the exchange.

For Redis, this is a no-op since Redis doesn't have real exchanges. We track exchange metadata for routing purposes.

bind

bind(channel: Channel) -> Exchange

Bind exchange to channel.

Queue

Queue

A Queue declaration.

Arguments: name: Name of the queue. Default is '' (auto-generated). exchange: The Exchange the queue binds to. routing_key: The binding key. Interpretation depends on exchange type. durable: Survive broker restart. Default is True. exclusive: Only consumable by current connection. Default is False. auto_delete: Delete when all consumers finished. Default is False. queue_arguments: Arguments for queue declare. binding_arguments: Arguments for queue bind. consumer_arguments: Arguments for consume. no_declare: Never declare this queue. expires: Queue expiry time in seconds. message_ttl: Message TTL in seconds. max_length: Maximum number of messages. max_length_bytes: Maximum total size in bytes. max_priority: Enable priority queue with max priority level. no_ack: Don't require acknowledgment. Default is False.

can_cache_declaration property

can_cache_declaration: bool

Whether one declaration holds for as long as the channel does.

The broker deletes an auto-delete queue when its last consumer goes and an expiring one once it sits unused, taking the binding with it, so those are declared every time, as upstream does. So is a queue bound to an exchange that can go away on its own.

from_dict classmethod

from_dict(
    name: str,
    *,
    exchange: Exchange | str | None = None,
    exchange_type: str | None = None,
    routing_key: str | None = None,
    **kwargs: Any,
) -> Queue

Create Queue from a dictionary of options.

This is used for backward compatibility when adding queues from config.

as_dict

as_dict(recurse: bool = False) -> dict[str, Any]

Return Queue as a dictionary.

Args: recurse: If True, also convert Exchange to dict.

declare async

declare(channel: Channel | None = None) -> str

Declare the exchange, the queue, and the binding between the two.

A queue on its own receives nothing on a broker with real exchanges, so declaring one means declaring all three.

Returns the queue name (useful for auto-generated names).

bind async

bind(channel: Channel | None = None) -> None

Bind the queue to its exchange.

get async

get(
    channel: Channel | None = None,
    no_ack: bool | None = None,
    accept: set[str] | None = None,
) -> Any

Get a single message from the queue.

Returns None if queue is empty.

purge async

purge(channel: Channel | None = None) -> int

Remove all messages from the queue.

Returns the number of messages deleted.

delete async

delete(
    channel: Channel | None = None,
    if_unused: bool = False,
    if_empty: bool = False,
) -> int

Delete the queue.

Returns the number of messages deleted.

bind_to_channel

bind_to_channel(channel: Channel) -> Queue

Bind queue to channel.

binding

binding

Represents a queue or exchange binding.

Arguments: exchange: Exchange to bind to. routing_key: Routing key used as binding key. arguments: Arguments for bind operation. unbind_arguments: Arguments for unbind operation.

declare async

declare(channel: Channel) -> None

Declare destination exchange.

bind async

bind(
    entity: Queue | Exchange, channel: Channel | None = None
) -> None

Bind entity to this binding.

unbind async

unbind(
    entity: Queue | Exchange, channel: Channel | None = None
) -> None

Unbind entity from this binding.