Channel

class amqpstorm.Channel[source]

RabbitMQ Channel.

e.g.

channel = connection.channel()
__init__(channel_id: int, connection: Connection, rpc_timeout: float) → None[source]
property basic: Basic

RabbitMQ Basic Operations.

e.g.

message = channel.basic.get(queue='hello_world')
Return type:

Basic

property exchange: Exchange

RabbitMQ Exchange Operations.

e.g.

channel.exchange.declare(exchange='hello_world')
Return type:

Exchange

property queue: Queue

RabbitMQ Queue Operations.

e.g.

channel.queue.declare(queue='hello_world')
Return type:

Queue

property tx: Tx

RabbitMQ Tx Operations.

e.g.

channel.tx.commit()
Return type:

Tx

build_inbound_messages(break_on_empty: bool = False, to_tuple: bool = False, auto_decode: bool = True, message_impl: type[BaseMessage] | None = None) → Iterator[Any][source]

Build messages in the inbound queue.

Parameters:

  • break_on_empty (bool) –

    Should we break the loop when there are no more messages in our inbound queue.

    To avoid breaking out while messages are still in flight from RabbitMQ, the loop only exits after the inbound queue has been continuously empty for at least one second.

  • to_tuple (bool) – Should incoming messages be converted to a tuple before delivery.

  • auto_decode (bool) – Auto-decode strings when possible.

  • message_impl (class) – Optional message class to use, derived from BaseMessage, for created messages. Defaults to Message.

Raises:
Return type:

generator

close(reply_code: int = 200, reply_text: str = '') → None[source]

Close Channel.

Parameters:

  • reply_code (int) – Close reply code (e.g. 200)

  • reply_text (str) – Close reply text

Raises:
Returns:

check_for_errors() → None[source]

Check connection and channel for errors.

Raises:
Returns:

check_for_exceptions() → None[source]

Check channel for exceptions.

Raises:

AMQPChannelError – Raises if the channel encountered an error.

Returns:

confirm_deliveries() → dict[str, Any][source]

Set the channel to confirm that each message has been successfully delivered.

Raises:
Returns:

property confirming_deliveries: bool

Is the channel set to confirm deliveries.

Returns:

process_data_events(to_tuple: bool = False, auto_decode: bool = True) → None[source]

Consume inbound messages.

Parameters:

  • to_tuple (bool) – Should incoming messages be converted to a tuple before delivery.

  • auto_decode (bool) – Auto-decode strings when possible.

Raises:
Returns:

start_consuming(to_tuple: bool = False, auto_decode: bool = True) → None[source]

Start consuming messages.

Parameters:

  • to_tuple (bool) – Should incoming messages be converted to a tuple before delivery.

  • auto_decode (bool) – Auto-decode strings when possible.

Raises:
Returns:

stop_consuming() → None[source]

Stop consuming messages.

Raises:
Returns:

classmethod __new__(*args, **kwargs)

Channel.Basic

class amqpstorm.basic.Basic[source]

RabbitMQ Basic Operations.

__init__(channel: Channel, max_frame_size: int | None = None) → None[source]
qos(prefetch_count: int = 0, prefetch_size: int = 0, global_: bool = False) → dict[str, Any][source]

Specify quality of service.

Parameters:

  • prefetch_count (int) – Prefetch window in messages

  • prefetch_size (int/long) – Prefetch window in octets

  • global (bool) – Apply to entire connection

Raises:
Return type:

dict

get(queue: str = '', no_ack: bool = False, to_dict: bool = False, auto_decode: bool = True, message_impl: type[BaseMessage] | None = None) → Message | BaseMessage | dict[str, Any] | None[source]

Fetch a single message.

Parameters:

  • queue (str) – Queue name

  • no_ack (bool) – No acknowledgement needed

  • to_dict (bool) – Should incoming messages be converted to a dictionary before delivery.

  • auto_decode (bool) – Auto-decode strings when possible.

  • message_impl (class) – Message implementation based on BaseMessage

Raises:
Returns:

Returns a single message, as long as there is a message in the queue. If no message is available, returns None.

Return type:

Message,dict,None

recover(requeue: bool = False) → dict[str, Any][source]

Redeliver unacknowledged messages.

Parameters:

requeue (bool) – Re-queue the messages

Raises:
Return type:

dict

consume(callback: Callable[[...], Any] | None = None, queue: str = '', consumer_tag: str = '', exclusive: bool = False, no_ack: bool = False, no_local: bool = False, arguments: dict[str, Any] | None = None) → str[source]

Start a queue consumer.

Parameters:

  • callback (Callable) – Message callback

  • queue (str) – Queue name

  • consumer_tag (str) – Consumer tag

  • no_local (bool) – Do not deliver own messages

  • no_ack (bool) – No acknowledgement needed

  • exclusive (bool) – Request exclusive access

  • arguments (dict) – Consume key/value arguments

Raises:
Returns:

Consumer tag

Return type:

str

cancel(consumer_tag: str = '') → dict[str, Any][source]

Cancel a queue consumer.

Parameters:

consumer_tag (str) – Consumer tag

Raises:
Return type:

dict

publish(body: bytes | str, routing_key: str, exchange: str = '', properties: dict[str, Any] | None = None, mandatory: bool = False, immediate: bool = False) → bool | None[source]

Publish a Message.

Parameters:

  • body (bytes,str,unicode) – Message payload

  • routing_key (str) – Message routing key

  • exchange (str) – The exchange to publish the message to

  • properties (dict) – Message properties

  • mandatory (bool) – Requires the message is published

  • immediate (bool) – Request immediate delivery

Raises:
Return type:

bool,None

ack(delivery_tag: int = 0, multiple: bool = False) → None[source]

Acknowledge Message.

Parameters:

  • delivery_tag (int/long) – Server-assigned delivery tag

  • multiple (bool) – Acknowledge multiple messages

Raises:
Returns:

nack(delivery_tag: int = 0, multiple: bool = False, requeue: bool = True) → None[source]

Negative Acknowledgement.

Parameters:

  • delivery_tag (int/long) – Server-assigned delivery tag

  • multiple (bool) – Negative acknowledge multiple messages

  • requeue (bool) – Re-queue the message

Raises:
Returns:

reject(delivery_tag: int = 0, requeue: bool = True) → None[source]

Reject Message.

Parameters:

  • delivery_tag (int/long) – Server-assigned delivery tag

  • requeue (bool) – Re-queue the message

Raises:
Returns:

Channel.Exchange

class amqpstorm.exchange.Exchange[source]

RabbitMQ Exchange Operations.

declare(exchange: str = '', exchange_type: str = 'direct', passive: bool = False, durable: bool = False, auto_delete: bool = False, arguments: dict[str, Any] | None = None) → dict[str, Any][source]

Declare an Exchange.

Parameters:

  • exchange (str) – Exchange name

  • exchange_type (str) – Exchange type

  • passive (bool) – Do not create

  • durable (bool) – Durable exchange

  • auto_delete (bool) – Automatically delete when not in use

  • arguments (dict) – Exchange key/value arguments

Raises:
Return type:

dict

delete(exchange: str = '', if_unused: bool = False) → dict[str, Any][source]

Delete an Exchange.

Parameters:

  • exchange (str) – Exchange name

  • if_unused (bool) – Delete only if unused

Raises:
Return type:

dict

bind(destination: str = '', source: str = '', routing_key: str = '', arguments: dict[str, Any] | None = None) → dict[str, Any][source]

Bind an Exchange.

Parameters:

  • destination (str) – Exchange name

  • source (str) – Exchange to bind to

  • routing_key (str) – The routing key to use

  • arguments (dict) – Bind key/value arguments

Raises:
Return type:

dict

unbind(destination: str = '', source: str = '', routing_key: str = '', arguments: dict[str, Any] | None = None) → dict[str, Any][source]

Unbind an Exchange.

Parameters:

  • destination (str) – Exchange name

  • source (str) – Exchange to unbind from

  • routing_key (str) – The routing key used

  • arguments (dict) – Unbind key/value arguments

Raises:
Return type:

dict

Channel.Queue

class amqpstorm.queue.Queue[source]

RabbitMQ Queue Operations.

declare(queue: str = '', passive: bool = False, durable: bool = False, exclusive: bool = False, auto_delete: bool = False, arguments: dict[str, Any] | None = None) → dict[str, Any][source]

Declare a Queue.

Parameters:

  • queue (str) – Queue name

  • passive (bool) – Do not create

  • durable (bool) – Durable queue

  • exclusive (bool) – Request exclusive access

  • auto_delete (bool) – Automatically delete when not in use

  • arguments (dict) – Queue key/value arguments

Raises:
Return type:

dict

delete(queue: str = '', if_unused: bool = False, if_empty: bool = False) → dict[str, Any][source]

Delete a Queue.

Parameters:

  • queue (str) – Queue name

  • if_unused (bool) – Delete only if unused

  • if_empty (bool) – Delete only if empty

Raises:
Return type:

dict

purge(queue: str) → dict[str, Any][source]

Purge a Queue.

Parameters:

queue (str) – Queue name

Raises:
Return type:

dict

bind(queue: str = '', exchange: str = '', routing_key: str = '', arguments: dict[str, Any] | None = None) → dict[str, Any][source]

Bind a Queue.

Parameters:

  • queue (str) – Queue name

  • exchange (str) – Exchange name

  • routing_key (str) – The routing key to use

  • arguments (dict) – Bind key/value arguments

Raises:
Return type:

dict

unbind(queue: str = '', exchange: str = '', routing_key: str = '', arguments: dict[str, Any] | None = None) → dict[str, Any][source]

Unbind a Queue.

Parameters:

  • queue (str) – Queue name

  • exchange (str) – Exchange name

  • routing_key (str) – The routing key used

  • arguments (dict) – Unbind key/value arguments

Raises:
Return type:

dict

Channel.Tx

class amqpstorm.tx.Tx[source]

RabbitMQ Transactions.

Server local transactions, in which the server will buffer published messages until the client commits (or rollback) the messages.

__init__(channel: Channel) → None[source]
select() → dict[str, Any] | None[source]

Enable standard transaction mode.

This will enable transaction mode on the channel. Meaning that messages will be kept in the remote server buffer until such a time that either commit or rollback is called.

Returns:

commit() → dict[str, Any] | None[source]

Commit the current transaction.

Commit all messages published during the current transaction session to the remote server.

A new transaction session starts as soon as the command has been executed.

Returns:

rollback() → dict[str, Any] | None[source]

Abandon the current transaction.

Rollback all messages published during the current transaction session to the remote server.

Note that all messages published during this transaction session will be lost, and will have to be published again.

A new transaction session starts as soon as the command has been executed.

Returns: