pika documentation - read the docs · pika documentation, release 0.9.9 pika is a pure-python...
TRANSCRIPT
pika DocumentationRelease 0.9.9
Gavin M. Roy
January 29, 2013
CONTENTS
i
ii
pika Documentation, Release 0.9.9
Pika is a pure-Python implementation of the AMQP 0-9-1 protocol that tries to stay fairly independent of the underly-ing network support library.
This documentation is a combination of both user documentation and module development documentation. Modulesand classes called out in the Using Pika section below will cover a majority of what users who are implementationpika in their applications will need. The Pika Core Objects section below lists all of the modules that are internal toPika.
If you have not developed with Pika or RabbitMQ before, the Connecting to RabbitMQ documentation is a good placeto get started.
CONTENTS 1
pika Documentation, Release 0.9.9
2 CONTENTS
CHAPTER
ONE
INSTALLING PIKA
Pika is available for download via PyPI and may be installed using easy_install or pip:
pip install pika
or:
easy_install pika
To install from source, run “python setup.py install” in the root source directory.
3
pika Documentation, Release 0.9.9
4 Chapter 1. Installing Pika
CHAPTER
TWO
USING PIKA
Pika supports two modes of development, synchronous using the BlockingConnection adapter and asynchronous usingone of the AsyncoreConnection, SelectConnection and TornadoConnection adapters.
2.1 Connecting to RabbitMQ
Pika provides multiple adapters to connect to RabbitMQ allowing for different ways of providing socket communica-tion depending on what is appropriate for your application.
• SelectConnection: A native event based connection adapter that implements select, kqueue, poll and epoll.
• AsyncoreConnection: Legacy adapter kept for convenience of previous Pika users.
• TornadoConnection: Connection adapter for use with the Tornado IO Loop.
• BlockingConnection: Enables blocking, synchronous operation on top of library for simple uses.
2.1.1 IO and Event Looping
Due to the need to check for and send content on a consistent basis, Pika now implements or extends IOLoops ineach of its asynchronous connection adapters. These IOLoops are blocking methods which loop and listen for events.Each asynchronous adapters follows the same standard for invoking the IOLoop. The IOLoop is created when theconnection adapter is created. To start it simply call the connection.ioloop.start() method.
If you are using an external IOLoop such as Tornado’s IOLoop, you may invoke that as you normally would and thenadd the adapter to it.
Example:
import pika
def on_open(connection):# Invoked when the connection is openpass
# Create our connection object, passing in the on_open methodconnection = pika.SelectConnection(on_open)
try:# Loop so we can communicate with RabbitMQconnection.ioloop.start()
except KeyboardInterrupt:# Gracefully close the connection
5
pika Documentation, Release 0.9.9
connection.close()# Loop until we’re fully closed, will stop on its ownconnection.ioloop.start()
2.1.2 Continuation-Passing Style
Interfacing with Pika asynchronously is done by passing in callback methods you would like to have invoked when acertain event has completed. For example, if you are going to declare a queue, you pass in a method that will be calledwhen the RabbitMQ server returns a Queue.DeclareOk response.
In our example below we use the following four easy steps:
1. We start by creating our connection object, then starting our event loop.
2. When we are connected, the on_connected method is called. In that method we create a channel.
3. When the channel is created, the on_channel_open method is called. In that method we declare a queue.
4. When the queue is declared successfully, on_queue_declared is called. In that method we callchannel.basic_consume telling it to call the handle_delivery for each message RabbitMQ delivers tous.
5. When RabbitMQ has a message to send us, it call the handle_delivery method passing the AMQP Method frame,Header frame and Body.
Note: Step #1 is on line #28 and Step #2 is on line #6. This is so that Python knows about the functions we’ll call inSteps #2 through #5.
Example:
import pika
# Create a global channel variable to hold our channel object inchannel = None
# Step #2def on_connected(connection):
"""Called when we are fully connected to RabbitMQ"""# Open a channelconnection.channel(on_channel_open)
# Step #3def on_channel_open(new_channel):
"""Called when our channel has opened"""global channelchannel = new_channelchannel.queue_declare(queue="test", durable=True, exclusive=False, auto_delete=False, callback=on_queue_declared)
# Step #4def on_queue_declared(frame):
"""Called when RabbitMQ has told us our Queue has been declared, frame is the response from RabbitMQ"""channel.basic_consume(handle_delivery, queue=’test’)
# Step #5def handle_delivery(channel, method, header, body):
"""Called when we receive a message from RabbitMQ"""print body
6 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
# Step #1: Connect to RabbitMQ using the default parametersparameters = pika.ConnectionParameters()connection = pika.SelectConnection(parameters, on_connected)
try:# Loop so we can communicate with RabbitMQconnection.ioloop.start()
except KeyboardInterrupt:# Gracefully close the connectionconnection.close()# Loop until we’re fully closed, will stop on its ownconnection.ioloop.start()
2.1.3 Credentials
The credentials module provides the mechanism by which you pass the username and password to theconnection.ConnectionParameters() class when it is created.
class pika.credentials.PlainCredentials(username, password, erase_on_connect=False)The PlainCredentials class returns the properly formatted username and password to the Connection. As ofthis version of Pika, only PlainCredentials are supported. To authenticate with Pika, simply create a credentialsobject passing in the username and password and pass that to the ConnectionParameters object.
If you do not pass in credentials to the ConnectionParameters object, it will create credentials for ‘guest’ withthe password of ‘guest’.
If you pass True to erase_on_connect the credentials will not be stored in memory after the Connection attempthas been made.
response_for(start)Validate that this type of authentication is supported
Parameters start (spec.Connection.Start) – Connection.Start method
Return type tuple(str|None, str|None)
erase_credentials()Called by Connection when it no longer needs the credentials
Example:
import pikacredentials = pika.PlainCredentials(’username’, ’password’)parameters = pika.ConnectionParameters(credentials=credentials)
2.1.4 Connection Parameters
There are two types of connection parameter classes in Pika to allow you to pass the connection information intoa connection adapter, ConnectionParameters and URLParameters. Both classes share the same default connectionvalues.
Default Parameter Values
The connection parameters classes extend pika.connection.Parameters to create a consistent definition of default valuesand internal attributes.
2.1. Connecting to RabbitMQ 7
pika Documentation, Release 0.9.9
pika.connection.Parameters = <class ‘pika.connection.Parameters’>Base connection parameters class definition
Parameters
• DEFAULT_HOST (str) – ‘localhost’
• DEFAULT_PORT (int) – 5672
• DEFAULT_VIRTUAL_HOST (str) – ‘/’
• DEFAULT_USERNAME (str) – ‘guest’
• DEFAULT_PASSWORD (str) – ‘guest’
• DEFAULT_HEARTBEAT_INTERVAL (int) – 0
• DEFAULT_CHANNEL_MAX (int) – 0
• DEFAULT_FRAME_MAX (int) – pika.spec.FRAME_MAX_SIZE
• DEFAULT_LOCALE (str) – ‘en_US’
• DEFAULT_CONNECTION_ATTEMPTS (int) – 1
• DEFAULT_RETRY_DELAY (int|float) – 2.0
• DEFAULT_SOCKET_TIMEOUT (int|float) – 0.25
• DEFAULT_SSL (bool) – False
• DEFAULT_SSL_OPTIONS (dict) – {}
• DEFAULT_SSL_PORT (int) – 5671
• DEFAULT_BACKPRESSURE_DETECTION (bool) – False
ConnectionParameters
The ConnectionParameters class allows you to specify the options needed when creating the object.
class pika.connection.ConnectionParameters(host=None, port=None, virtual_host=None,credentials=None, channel_max=None,frame_max=None, heartbeat_interval=None,ssl=None, ssl_options=None, connec-tion_attempts=None, retry_delay=None,socket_timeout=None, locale=None, back-pressure_detection=None)
Connection parameters object that is passed into the connection adapter upon construction.
Example:
import pika
# Set the connection parameters to connect to rabbit-server1 on port 5672# on the / virtual host using the username "guest" and password "guest"credentials = pika.PlainCredentials(’guest’, ’guest’)parameters = pika.ConnectionParameters(’rabbit-server1’,
5672’/’,credentials)
8 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
URLParameters
The URLParameters class allows you to pass in an AMQP URL when creating the object and supports the host, port,virtual host, ssl, username and password in the base URL and other options are passed in via query parameters.
class pika.connection.URLParameters(url)Create a Connection parameters object based off of URIParameters
Example:
import pika
# Set the connection parameters to connect to rabbit-server1 on port 5672# on the / virtual host using the username "guest" and password "guest"parameters = pika.URLParameters(’amqp://guest:guest@rabbit-server1:5672/%2F’)
2.1.5 TCP Backpressure
As of RabbitMQ 2.0, client side Channel.Flow has been removed 1. Instead, the RabbitMQ broker uses TCP Back-pressure to slow your client if it is delivering messages too fast. If you pass in backpressure_detection into yourconnection parameters, Pika attempts to help you handle this situation by providing a mechanism by which you maybe notified if Pika has noticed too many frames have yet to be delivered. By registering a callback function with theadd_backpressure_callback method of any connection adapter, your function will be called when Pika seesthat a backlog of 10 times the average frame size you have been sending has been exceeded. You may tweak thenotification multiplier value by calling the set_backpressure_multiplier method passing any integer value.
Example:
import pika
parameters = pika.URLParameters(’amqp://guest:guest@rabbit-server1:5672/%2F?backpressure_detection=t’)
2.1.6 Available Adapters
The following connection adapters are available for connecting with RabbitMQ:
AsyncoreConnection
Note: Use It is recommended that you use SelectConnection and its method signatures are the same as Asyncore-Connection.
The AsyncoreConnection class is provided for legacy support and quicker porting from applications that used Pikaversion 0.5.2 and prior.
class pika.adapters.asyncore_connection.AsyncoreConnection(parameters=None,on_open_callback=None,stop_ioloop_on_close=True)
add_backpressure_callback(callback_method)Call method “callback” when pika believes backpressure is being applied.
1 “more effective flow control mechanism that does not require cooperation from clients and reacts quickly to prevent the bro-ker from exhausing memory - see http://www.rabbitmq.com/extensions.html#memsup” from http://lists.rabbitmq.com/pipermail/rabbitmq-announce/attachments/20100825/2c672695/attachment.txt
2.1. Connecting to RabbitMQ 9
pika Documentation, Release 0.9.9
Parameters callback_method (method) – The method to call
add_on_close_callback(callback_method)Add a callback notification when the connection has closed.
Parameters callback_method (method) – The callback when the channel is opened
add_on_open_callback(callback_method)Add a callback notification when the connection has opened.
Parameters callback_method (method) – The callback when the channel is opened
add_timeout(deadline, callback_method)Add the callback_method to the IOLoop timer to fire after deadline seconds. Returns a handle to thetimeout
Parameters
• deadline (int) – The number of seconds to wait to call callback
• callback_method (method) – The callback method
Return type str
basic_nackSpecifies if the server supports basic.nack on the active connection.
Return type bool
channel(on_open_callback, channel_number=None)Create a new channel with the next available channel number or pass in a channel number to use. Must benon-zero if you would like to specify but it is recommended that you let Pika manage the channel numbers.
Parameters
• on_open_callback (method) – The callback when the channel is opened
• channel_number (int) – The channel number to use, defaults to the next available.
Return type pika.channel.Channel
close(reply_code=200, reply_text=’Normal shutdown’)Disconnect from RabbitMQ. If there are any open channels, it will attempt to close them prior to fullydisconnecting. Channels which have active consumers will attempt to send a Basic.Cancel to RabbitMQto cleanly stop the delivery of messages prior to closing the channel.
Parameters
• reply_code (int) – The code number for the close
• reply_text (str) – The text reason for the close
consumer_cancel_notifySpecifies if the server supports consumer cancel notification on the active connection.
Return type bool
exchange_exchange_bindingsSpecifies if the active connection supports exchange to exchange bindings.
Return type bool
is_closedReturns a boolean reporting the current connection state.
is_closingReturns a boolean reporting the current connection state.
10 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
is_openReturns a boolean reporting the current connection state.
publisher_confirmsSpecifies if the active connection can use publisher confirmations.
Return type bool
remove_timeout(timeout_id)Remove the timeout from the IOLoop by the ID returned from add_timeout.
Return type str
set_backpressure_multiplier(value=10)Alter the backpressure multiplier value. We set this to 10 by default. This value is used to raise warningsand trigger the backpressure callback.
Parameters value (int) – The multiplier value to set
BlockingConnection
The BlockingConnection creates a layer on top of Pika’s asynchronous core providng methods that will block untiltheir expected response has returned. Due to the asynchronous nature of the Basic.Deliver and Basic.Return calls fromRabbitMQ to your application, you are still required to implement continuation-passing style asynchronous methods ifyou’d like to receive messages from RabbitMQ using basic_consume or if you want to be notified of a delivery failurewhen using basic_publish.
Basic.Get is a blocking call which will either return the Method Frame, Header Frame and Body of a message, or itwill return a Basic.GetEmpty frame as the Method Frame.
For more information on using the BlockingConnection, see BlockingChannel
Publishing Example:
import pika
# Open a connection to RabbitMQ on localhost using all default parametersconnection = pika.BlockingConnection()
# Open the channelchannel = connection.channel()
# Declare the queuechannel.queue_declare(queue="test", durable=True, exclusive=False, auto_delete=False)
# Send a messagechannel.basic_publish(exchange=’’,
routing_key="test",body="Hello World!",properties=pika.BasicProperties(content_type="text/plain",
delivery_mode=1))
Consuming Example:
import pika
# Open a connection to RabbitMQ on localhost using all default parametersconnection = pika.BlockingConnection()
# Open the channel
2.1. Connecting to RabbitMQ 11
pika Documentation, Release 0.9.9
channel = connection.channel()
# Declare the queuechannel.queue_declare(queue="test", durable=True,
exclusive=False, auto_delete=False)
# Start our counter at 0messages = 0
# Method that will receive our messages and stop consuming after 10def _on_message(channel, method, header, body):
print "Message:"print "\t%r" % methodprint "\t%r" % headerprint "\t%r" % body
# Acknowledge message receiptchannel.basic_ack(method.delivery_tag)
# We’ve received 10 messages, stop consumingglobal messagesmessages += 1if messages > 10:
channel.stop_consuming()
# Setup up our consumer callbackchannel.basic_consume(_on_message, queue="test")
# This is blocking until channel.stop_consuming is called and will allow us to receive messageschannel.start_consuming()
Implement a blocking, procedural style connection adapter on top of the asynchronous core.
class pika.adapters.blocking_connection.BlockingConnection(parameters=None,on_open_callback=None,stop_ioloop_on_close=True)
The BlockingConnection adapter is meant for simple implementations where you want to have blocking behav-ior. The behavior layered on top of the async library. Because of the nature of AMQP there are a few callbacksone needs to do, even in a blocking implementation. These include receiving messages from Basic.Deliver,Basic.GetOk, and Basic.Return.
add_timeout(deadline, callback)Add the callback to the IOLoop timer to fire after deadline seconds.
Parameters
• deadline (int) – The number of seconds to wait to call callback
• callback (method) – The callback method
Return type str
channel(channel_number=None)Create a new channel with the next available or specified channel #.
Parameters channel_number (int) – Specify the channel number
close(reply_code=200, reply_text=’Normal shutdown’)Disconnect from RabbitMQ. If there are any open channels, it will attempt to close them prior to fullydisconnecting. Channels which have active consumers will attempt to send a Basic.Cancel to RabbitMQto cleanly stop the delivery of messages prior to closing the channel.
12 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
Parameters
• reply_code (int) – The code number for the close
• reply_text (str) – The text reason for the close
disconnect()Disconnect from the socket
process_data_events()Will make sure that data events are processed. Your app can block on this method.
process_timeouts()Process the self._timeouts event stack
remove_timeout(timeout_id)Remove the timeout from the IOLoop by the ID returned from add_timeout.
Parameters timeout_id (str) – The id of the timeout to remove
send_method(channel_number, method_frame, content=None)Constructs a RPC method frame and then sends it to the broker.
Parameters
• channel_number (int) – The channel number for the frame
• method_frame (pika.object.Method) – The method frame to send
• content (tuple) – If set, is a content frame, is tuple of properties and body.
sleep(duration)A safer way to sleep than calling time.sleep() directly which will keep the adapter from ignoring framessent from RabbitMQ. The connection will “sleep” or block the number of seconds specified in duration insmall intervals.
Parameters duration (int) – The time to sleep
add_backpressure_callback(callback_method)Call method “callback” when pika believes backpressure is being applied.
Parameters callback_method (method) – The method to call
add_on_close_callback(callback_method)Add a callback notification when the connection has closed.
Parameters callback_method (method) – The callback when the channel is opened
add_on_open_callback(callback_method)Add a callback notification when the connection has opened.
Parameters callback_method (method) – The callback when the channel is opened
basic_nackSpecifies if the server supports basic.nack on the active connection.
Return type bool
consumer_cancel_notifySpecifies if the server supports consumer cancel notification on the active connection.
Return type bool
exchange_exchange_bindingsSpecifies if the active connection supports exchange to exchange bindings.
Return type bool
2.1. Connecting to RabbitMQ 13
pika Documentation, Release 0.9.9
is_closedReturns a boolean reporting the current connection state.
is_closingReturns a boolean reporting the current connection state.
is_openReturns a boolean reporting the current connection state.
publisher_confirmsSpecifies if the active connection can use publisher confirmations.
Return type bool
set_backpressure_multiplier(value=10)Alter the backpressure multiplier value. We set this to 10 by default. This value is used to raise warningsand trigger the backpressure callback.
Parameters value (int) – The multiplier value to set
SelectConnection
Note: SelectConnection is the recommended method for using Pika under most circumstances. It supports multipleevent notification methods including select, epoll, kqueue and poll.
By default SelectConnection will attempt to use the most appropriate event notification method for your system. Inorder to override the default behavior you may set the poller type by assigning a string value to the select_connectionmodules POLLER_TYPE attribute prior to creating the SelectConnection object instance. Valid values are: kqueue,poll, epoll, select
Poller Type Override Example:
from pika.adapters import select_connection
select_connection.POLLER_TYPE = ’epoll’connection = select_connection.SelectConnection()
See the Continuation-Passing Style example for an example of using SelectConnection. A connection adapter thattries to use the best polling method for the platform pika is running on.
class pika.adapters.select_connection.SelectConnection(parameters=None,on_open_callback=None,stop_ioloop_on_close=True)
An asynchronous connection adapter that attempts to use the fastest event loop adapter for the given platform.
add_backpressure_callback(callback_method)Call method “callback” when pika believes backpressure is being applied.
Parameters callback_method (method) – The method to call
add_on_close_callback(callback_method)Add a callback notification when the connection has closed.
Parameters callback_method (method) – The callback when the channel is opened
add_on_open_callback(callback_method)Add a callback notification when the connection has opened.
Parameters callback_method (method) – The callback when the channel is opened
14 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
add_timeout(deadline, callback_method)Add the callback_method to the IOLoop timer to fire after deadline seconds. Returns a handle to thetimeout
Parameters
• deadline (int) – The number of seconds to wait to call callback
• callback_method (method) – The callback method
Return type str
basic_nackSpecifies if the server supports basic.nack on the active connection.
Return type bool
channel(on_open_callback, channel_number=None)Create a new channel with the next available channel number or pass in a channel number to use. Must benon-zero if you would like to specify but it is recommended that you let Pika manage the channel numbers.
Parameters
• on_open_callback (method) – The callback when the channel is opened
• channel_number (int) – The channel number to use, defaults to the next available.
Return type pika.channel.Channel
close(reply_code=200, reply_text=’Normal shutdown’)Disconnect from RabbitMQ. If there are any open channels, it will attempt to close them prior to fullydisconnecting. Channels which have active consumers will attempt to send a Basic.Cancel to RabbitMQto cleanly stop the delivery of messages prior to closing the channel.
Parameters
• reply_code (int) – The code number for the close
• reply_text (str) – The text reason for the close
consumer_cancel_notifySpecifies if the server supports consumer cancel notification on the active connection.
Return type bool
exchange_exchange_bindingsSpecifies if the active connection supports exchange to exchange bindings.
Return type bool
is_closedReturns a boolean reporting the current connection state.
is_closingReturns a boolean reporting the current connection state.
is_openReturns a boolean reporting the current connection state.
publisher_confirmsSpecifies if the active connection can use publisher confirmations.
Return type bool
remove_timeout(timeout_id)Remove the timeout from the IOLoop by the ID returned from add_timeout.
2.1. Connecting to RabbitMQ 15
pika Documentation, Release 0.9.9
Return type str
set_backpressure_multiplier(value=10)Alter the backpressure multiplier value. We set this to 10 by default. This value is used to raise warningsand trigger the backpressure callback.
Parameters value (int) – The multiplier value to set
TornadoConnection
Tornado is an open source version of the scalable, non-blocking web server and tools that power FriendFeed. For moreinformation on tornado, visit http://tornadoweb.org
Since the Tornado IOLoop blocks once it is started, it is suggested that you use a timer to add Pika to your tor-nado.Application instance after the HTTPServer has started.
The following is a simple, non-working example on how to add Pika to the Tornado IOLoop without blocking otherapplications from doing so. To see a fully workng example, see the Tornado Demo application in the examples.
Example:
from pika.adapters.tornado_connection import TornadoConnection
class PikaClient(object):def connect(self):
self.connection = TornadoConnection(on_connected_callback=self.on_connected)
# Create our Tornado Applicationapplication = tornado.web.Application([
(r"/", ExampleHandler)], **settings)
# Create our Pika Clientapplication.pika = PikaClient()
# Start the HTTPServerhttp_server = tornado.httpserver.HTTPServer(application)http_server.listen(8080)
# Get a handle to the instance of IOLoopioloop = tornado.ioloop.IOLoop.instance()
# Add our Pika connect to the IOLoop since we loop on ioloop.startioloop.add_timeout(500, application.pika.connect)
# Start the IOLoopioloop.start()
Run pika on the Tornado IOLoop
class pika.adapters.tornado_connection.TornadoConnection(parameters=None,on_open_callback=None,stop_ioloop_on_close=False,custom_ioloop=None)
The TornadoConnection runs on the Tornado IOLoop. If you’re running the connection in a web app, make sureyou set stop_ioloop_on_close to False, which is the default behavior for this adapter, otherwise the web app willstop taking requests.
add_timeout(deadline, callback_method)Add the callback_method to the IOLoop timer to fire after deadline seconds. Returns a handle to the
16 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
timeout. Do not confuse with Tornado’s timeout where you pass in the time you want to have your callbackcalled. Only pass in the seconds until it’s to be called.
Parameters
• deadline (int) – The number of seconds to wait to call callback
• callback_method (method) – The callback method
Return type str
remove_timeout(timeout_id)Remove the timeout from the IOLoop by the ID returned from add_timeout.
Return type str
add_backpressure_callback(callback_method)Call method “callback” when pika believes backpressure is being applied.
Parameters callback_method (method) – The method to call
add_on_close_callback(callback_method)Add a callback notification when the connection has closed.
Parameters callback_method (method) – The callback when the channel is opened
add_on_open_callback(callback_method)Add a callback notification when the connection has opened.
Parameters callback_method (method) – The callback when the channel is opened
basic_nackSpecifies if the server supports basic.nack on the active connection.
Return type bool
channel(on_open_callback, channel_number=None)Create a new channel with the next available channel number or pass in a channel number to use. Must benon-zero if you would like to specify but it is recommended that you let Pika manage the channel numbers.
Parameters
• on_open_callback (method) – The callback when the channel is opened
• channel_number (int) – The channel number to use, defaults to the next available.
Return type pika.channel.Channel
close(reply_code=200, reply_text=’Normal shutdown’)Disconnect from RabbitMQ. If there are any open channels, it will attempt to close them prior to fullydisconnecting. Channels which have active consumers will attempt to send a Basic.Cancel to RabbitMQto cleanly stop the delivery of messages prior to closing the channel.
Parameters
• reply_code (int) – The code number for the close
• reply_text (str) – The text reason for the close
consumer_cancel_notifySpecifies if the server supports consumer cancel notification on the active connection.
Return type bool
exchange_exchange_bindingsSpecifies if the active connection supports exchange to exchange bindings.
2.1. Connecting to RabbitMQ 17
pika Documentation, Release 0.9.9
Return type bool
is_closedReturns a boolean reporting the current connection state.
is_closingReturns a boolean reporting the current connection state.
is_openReturns a boolean reporting the current connection state.
publisher_confirmsSpecifies if the active connection can use publisher confirmations.
Return type bool
set_backpressure_multiplier(value=10)Alter the backpressure multiplier value. We set this to 10 by default. This value is used to raise warningsand trigger the backpressure callback.
Parameters value (int) – The multiplier value to set
2.2 Using the Channel Object
Channels are the main communication mechanism in pika for interacting with RabbitMQ. Pika automatically choosesthe appropriate channel object for you to use when you create one with a connection adapter:
Example of creating a Channel:
import pika
# Create our connection object connection = pika.SelectConnection(on_channel_open)
def on_open(connection):
# The returned object will be a blocking channel channel = connec-tion.channel(on_channel_open)
def on_channel_open(channel):
# It is also passed into the event that is fired when the channel is opened print channel
When using the BlockingConnection adapter, the channel object returned is a BlockingChannel.
2.2.1 Channel
The Channel class provides a wrapper for interacting with RabbitMQ implementing the methods and behaviors for anAMQP Channel.
class pika.channel.Channel(connection, channel_number, on_open_callback=None)A Channel is the primary communication method for interacting with RabbitMQ. It is recommended that youdo not directly invoke the creation of a channel object in your application code but rather construct the a channelby calling the active connection’s channel() method.
add_callback(callback, replies, one_shot=True)Pass in a callback handler and a list replies from the RabbitMQ broker which you’d like the callbacknotified of. Callbacks should allow for the frame parameter to be passed in.
Parameters
• callback (method) – The method to call
18 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
• replies (list) – The replies to get a callback for
• one_shot (bool) – Only handle the first type callback
add_on_cancel_callback(callback)Pass a callback function that will be called when the basic_cancel is sent by the server. The callbackfunction should receive a frame parameter.
Parameters callback (method) – The method to call on callback
add_on_close_callback(callback)Pass a callback function that will be called when the channel is closed. The callback function shouldreceive a frame parameter.
Parameters callback (method) – The method to call on callback
add_on_flow_callback(callback)Pass a callback function that will be called when Channel.Flow is called by the remote server. Note thatnewer versions of RabbitMQ will not issue this but instead use TCP backpressure
Parameters callback (method) – The method to call on callback
add_on_return_callback(callback)Pass a callback function that will be called when basic_publish as sent a message that has been rejectedand returned by the server. The callback handler should receive a method, header and body frame. Thebase signature for the callback should be the same as the method signature one creates for a basic_consumecallback.
Parameters callback (method) – The method to call on callback
basic_ack(delivery_tag=0, multiple=False)Acknowledge one or more messages. When sent by the client, this method acknowledges one or more mes-sages delivered via the Deliver or Get-Ok methods. When sent by server, this method acknowledges one ormore messages published with the Publish method on a channel in confirm mode. The acknowledgementcan be for a single message or a set of messages up to and including a specific message.
Parameters
• delivery-tag (int) – The server-assigned delivery tag
• multiple (bool) – If set to True, the delivery tag is treated as “up to and including”, so thatmultiple messages can be acknowledged with a single method. If set to False, the deliverytag refers to a single message. If the multiple field is 1, and the delivery tag is zero, thisindicates acknowledgement of all outstanding messages.
basic_cancel(callback=None, consumer_tag=’‘, nowait=False)This method cancels a consumer. This does not affect already delivered messages, but it does mean theserver will not send any more messages for that consumer. The client may receive an arbitrary numberof messages in between sending the cancel method and receiving the cancel-ok reply. It may also be sentfrom the server to the client in the event of the consumer being unexpectedly cancelled (i.e. cancelled forany reason other than the server receiving the corresponding basic.cancel from the client). This allowsclients to be notified of the loss of consumers due to events such as queue deletion.
Parameters
• callback (method) – Method to call for a Basic.CancelOk response
• consumer_tag (str) – Identifier for the consumer
• nowait (bool) – Do not expect a Basic.CancelOk response
Raises ValueError
2.2. Using the Channel Object 19
pika Documentation, Release 0.9.9
basic_consume(consumer_callback, queue=’‘, no_ack=False, exclusive=False, con-sumer_tag=None)
Sends the AMQP command Basic.Consume to the broker and binds messages for the consumer_tag to theconsumer callback. If you do not pass in a consumer_tag, one will be automatically generated for you.Returns the consumer tag.
For more information on basic_consume, see: http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.consume
Parameters
• consumer_callback (method) – The method to callback when consuming
• queue (str or unicode) – The queue to consume from
• no_ack (bool) – Tell the broker to not expect a response
• exclusive (bool) – Don’t allow other consumers on the queue
• consumer_tag (str or unicode) – Specify your own consumer tag
Return type str
basic_get(callback=None, queue=’‘, no_ack=False)Get a single message from the AMQP broker. The callback method signature should have 3 parameters:The method frame, header frame and the body, like the consumer callback for Basic.Consume. If you wantto be notified of Basic.GetEmpty, use the Channel.add_callback method adding your Basic.GetEmptycallback which should expect only one parameter, frame. For more information on basic_get and itsparameters, see:
http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.get
Parameters
• callback (method) – The method to callback with a message
• queue (str or unicode) – The queue to get a message from
• no_ack (bool) – Tell the broker to not expect a reply
basic_nack(delivery_tag=None, multiple=False, requeue=True)This method allows a client to reject one or more incoming messages. It can be used to interrupt and cancellarge incoming messages, or return untreatable messages to their original queue.
Parameters
• delivery-tag (int) – The server-assigned delivery tag
• multiple (bool) – If set to True, the delivery tag is treated as “up to and including”, so thatmultiple messages can be acknowledged with a single method. If set to False, the deliverytag refers to a single message. If the multiple field is 1, and the delivery tag is zero, thisindicates acknowledgement of all outstanding messages.
• requeue (bool) – If requeue is true, the server will attempt to requeue the message. Ifrequeue is false or the requeue attempt fails the messages are discarded or dead-lettered.
basic_publish(exchange, routing_key, body, properties=None, mandatory=False, immedi-ate=False)
Publish to the channel with the given exchange, routing key and body. For more information on ba-sic_publish and what the parameters do, see:
http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.publish
Parameters
• exchange (str or unicode) – The exchange to publish to
20 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
• routing_key (str or unicode) – The routing key to bind on
• body (str or unicode) – The message body
• properties (pika.spec.Properties) – Basic.properties
• mandatory (bool) – The mandatory flag
• immediate (bool) – The immediate flag
basic_qos(callback=None, prefetch_size=0, prefetch_count=0, all_channels=False)Specify quality of service. This method requests a specific quality of service. The QoS can be specifiedfor the current channel or for all channels on the connection. The client can request that messages be sentin advance so that when the client finishes processing a message, the following message is already heldlocally, rather than needing to be sent down the channel. Prefetching gives a performance improvement.
Parameters
• callback (method) – The method to callback for Basic.QosOk response
• prefetch_size (int) – This field specifies the prefetch window size. The server will senda message in advance if it is equal to or smaller in size than the available prefetch size(and also falls into other prefetch limits). May be set to zero, meaning “no specific limit”,although other prefetch limits may still apply. The prefetch-size is ignored if the no-ackoption is set.
• prefetch_count (int) – Specifies a prefetch window in terms of whole messages. Thisfield may be used in combination with the prefetch-size field; a message will only be sentin advance if both prefetch windows (and those at the channel and connection level) allowit. The prefetch-count is ignored if the no-ack option is set.
• all_channels (bool) – Should the QoS apply to all channels
basic_recover(callback=None, requeue=False)This method asks the server to redeliver all unacknowledged messages on a specified channel. Zero ormore messages may be redelivered. This method replaces the asynchronous Recover.
Parameters
• callback (method) – Method to call when receiving Basic.RecoverOk
• requeue (bool) – If False, the message will be redelivered to the original recipient. IfTrue, the server will attempt to requeue the message, potentially then delivering it to analternative subscriber.
basic_reject(delivery_tag=None, requeue=True)Reject an incoming message. This method allows a client to reject a message. It can be used to interruptand cancel large incoming messages, or return untreatable messages to their original queue.
Parameters
• delivery-tag (int) – The server-assigned delivery tag
• requeue (bool) – If requeue is true, the server will attempt to requeue the message. Ifrequeue is false or the requeue attempt fails the messages are discarded or dead-lettered.
close(reply_code=0, reply_text=’Normal Shutdown’)Will invoke a clean shutdown of the channel with the AMQP Broker.
Parameters
• reply_code (int) – The reply code to close the channel with
• reply_text (str) – The reply text to close the channel with
2.2. Using the Channel Object 21
pika Documentation, Release 0.9.9
confirm_delivery(callback=None, nowait=False)Turn on Confirm mode in the channel. Pass in a callback to be notified by the Broker when a message hasbeen confirmed as received or rejected (Basic.Ack, Basic.Nack) from the broker to the publisher.
For more information see: http://www.rabbitmq.com/extensions.html#confirms
Parameters
• callback (method) – The callback for delivery confirmations
• nowait (bool) – Do not send a reply frame (Confirm.SelectOk)
consumer_tagsProperty method that returns a list of currently active consumers
Return type list
exchange_bind(callback=None, destination=None, source=None, routing_key=’‘, nowait=False, ar-guments=None)
Bind an exchange to another exchange.
Parameters
• callback (method) – The method to call on Exchange.BindOk
• destination (str or unicode) – The destination exchange to bind
• source (str or unicode) – The source exchange to bind to
• routing_key (str or unicode) – The routing key to bind on
• nowait (bool) – Do not wait for an Exchange.BindOk
• arguments (dict) – Custom key/value pair arguments for the binding
exchange_declare(callback=None, exchange=None, exchange_type=’direct’, passive=False,durable=False, auto_delete=False, internal=False, nowait=False, argu-ments=None, type=None)
This method creates an exchange if it does not already exist, and if the exchange exists, verifies that it isof the correct and expected class.
If passive set, the server will reply with Declare-Ok if the exchange already exists with the same name, andraise an error if not and if the exchange does not already exist, the server MUST raise a channel exceptionwith reply code 404 (not found).
Parameters
• callback (method) – Call this method on Exchange.DeclareOk
• exchange (str or unicode sequence of these characters: letters, digits, hyphen, underscore,period, or colon.) – The exchange name consists of a non-empty
• exchange_type (str) – The exchange type to use
• passive (bool) – Perform a declare or just check to see if it exists
• durable (bool) – Survive a reboot of RabbitMQ
• auto_delete (bool) – Remove when no more queues are bound to it
• internal (bool) – Can only be published to by other exchanges
• nowait (bool) – Do not expect an Exchange.DeclareOk response
• arguments (dict) – Custom key/value pair arguments for the exchange
• type (str) – The deprecated exchange type parameter
22 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
exchange_delete(callback=None, exchange=None, if_unused=False, nowait=False)Delete the exchange.
Parameters
• callback (method) – The method to call on Exchange.DeleteOk
• exchange (str or unicode) – The exchange name
• if_unused (bool) – only delete if the exchange is unused
• nowait (bool) – Do not wait for an Exchange.DeleteOk
exchange_unbind(callback=None, destination=None, source=None, routing_key=’‘, nowait=False,arguments=None)
Unbind an exchange from another exchange.
Parameters
• callback (method) – The method to call on Exchange.UnbindOk
• destination (str or unicode) – The destination exchange to unbind
• source (str or unicode) – The source exchange to unbind from
• routing_key (str or unicode) – The routing key to unbind
• nowait (bool) – Do not wait for an Exchange.UnbindOk
• arguments (dict) – Custom key/value pair arguments for the binding
flow(callback, active)Turn Channel flow control off and on. Pass a callback to be notified of the response from the server. activeis a bool. Callback should expect a bool in response indicating channel flow state. For more information,please reference:
http://www.rabbitmq.com/amqp-0-9-1-reference.html#channel.flow
Parameters
• callback (method) – The callback method
• active (bool) – Turn flow on or off
is_closedReturns True if the channel is closed.
Return type bool
is_closingReturns True if the channel is closing.
Return type bool
is_openReturns True if the channel is open.
Return type bool
open()Open the channel
queue_bind(callback, queue, exchange, routing_key=None, nowait=False, arguments=None)Bind the queue to the specified exchange
Parameters
• callback (method) – The method to call on Queue.BindOk
2.2. Using the Channel Object 23
pika Documentation, Release 0.9.9
• queue (str or unicode) – The queue to bind to the exchange
• exchange (str or unicode) – The source exchange to bind to
• routing_key (str or unicode) – The routing key to bind on
• nowait (bool) – Do not wait for a Queue.BindOk
• arguments (dict) – Custom key/value pair arguments for the binding
queue_declare(callback, queue=’‘, passive=False, durable=False, exclusive=False,auto_delete=False, nowait=False, arguments=None)
Declare queue, create if needed. This method creates or checks a queue. When creating a new queue theclient can specify various properties that control the durability of the queue and its contents, and the levelof sharing for the queue.
Leave the queue name empty for a auto-named queue in RabbitMQ
Parameters
• callback (method) – The method to call on Queue.DeclareOk
• queue (str or unicode) – The queue name
• passive (bool) – Only check to see if the queue exists
• durable (bool) – Survive reboots of the broker
• exclusive (bool) – Only allow access by the current connection
• auto_delete (bool) – Delete after consumer cancels or disconnects
• nowait (bool) – Do not wait for a Queue.DeclareOk
• arguments (dict) – Custom key/value arguments for the queue
queue_delete(callback=None, queue=’‘, if_unused=False, if_empty=False, nowait=False)Delete a queue from the broker.
Parameters
• callback (method) – The method to call on Queue.DeleteOk
• queue (str or unicode) – The queue to delete
• if_unused (bool) – only delete if it’s unused
• if_empty (bool) – only delete if the queue is empty
• nowait (bool) – Do not wait for a Queue.DeleteOk
queue_purge(callback=None, queue=’‘, nowait=False)Purge all of the messages from the specified queue
Parameters
• callback (method) – The method to call on Queue.PurgeOk
• queue (str or unicode) – The queue to purge
• nowait (bool) – Do not expect a Queue.PurgeOk response
queue_unbind(callback=None, queue=’‘, exchange=None, routing_key=’‘, arguments=None)Unbind a queue from an exchange.
Parameters
• callback (method) – The method to call on Queue.UnbindOk
• queue (str or unicode) – The queue to unbind from the exchange
24 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
• exchange (str or unicode) – The source exchange to bind from
• routing_key (str or unicode) – The routing key to unbind
• arguments (dict) – Custom key/value pair arguments for the binding
tx_commit(callback=None)Commit a transaction
Parameters callback (method) – The callback for delivery confirmations
tx_rollback(callback=None)Rollback a transaction.
Parameters callback (method) – The callback for delivery confirmations
tx_select(callback=None)Select standard transaction mode. This method sets the channel to use standard transactions. The clientmust use this method at least once on a channel before using the Commit or Rollback methods.
Parameters callback (method) – The callback for delivery confirmations
2.3 Using the BlockingChannel
Channels are the main communication mechanism in pika for interacting with RabbitMQ. Pika automatically choosesthe appropriate channel object for you to use when you create one with a connection adapter.
Example of creating a BlockingChannel:
import pika
# Create our connection objectconnection = pika.BlockingConnection()
# The returned object will be a blocking channelchannel = connection.channel()
The BlockingChannel implements blocking semantics for most things that one would use callback-passing-style forwith the Channel class. In addition, the BlockingChannel class implements a generator that allows you to consumemessages without using callbacks.
2.3.1 BlockingChannel
Implement a blocking, procedural style connection adapter on top of the asynchronous core.
class pika.adapters.blocking_connection.BlockingChannel(connection, channel_number)The BlockingChannel class implements a blocking layer on top of the Channel class.
add_callback(callback, replies, one_shot=True)Pass in a callback handler and a list replies from the RabbitMQ broker which you’d like the callbacknotified of. Callbacks should allow for the frame parameter to be passed in.
Parameters
• callback (method) – The method to call
• replies (list) – The replies to get a callback for
• one_shot (bool) – Only handle the first type callback
2.3. Using the BlockingChannel 25
pika Documentation, Release 0.9.9
add_on_cancel_callback(callback)Pass a callback function that will be called when the basic_cancel is sent by the server. The callbackfunction should receive a frame parameter.
Parameters callback (method) – The method to call on callback
add_on_close_callback(callback)Pass a callback function that will be called when the channel is closed. The callback function shouldreceive a frame parameter.
Parameters callback (method) – The method to call on callback
add_on_flow_callback(callback)Pass a callback function that will be called when Channel.Flow is called by the remote server. Note thatnewer versions of RabbitMQ will not issue this but instead use TCP backpressure
Parameters callback (method) – The method to call on callback
add_on_return_callback(callback)Pass a callback function that will be called when basic_publish as sent a message that has been rejectedand returned by the server. The callback handler should receive a method, header and body frame. Thebase signature for the callback should be the same as the method signature one creates for a basic_consumecallback.
Parameters callback (method) – The method to call on callback
basic_ack(delivery_tag=0, multiple=False)Acknowledge one or more messages. When sent by the client, this method acknowledges one or more mes-sages delivered via the Deliver or Get-Ok methods. When sent by server, this method acknowledges one ormore messages published with the Publish method on a channel in confirm mode. The acknowledgementcan be for a single message or a set of messages up to and including a specific message.
Parameters
• delivery-tag (int) – The server-assigned delivery tag
• multiple (bool) – If set to True, the delivery tag is treated as “up to and including”, so thatmultiple messages can be acknowledged with a single method. If set to False, the deliverytag refers to a single message. If the multiple field is 1, and the delivery tag is zero, thisindicates acknowledgement of all outstanding messages.
basic_cancel(consumer_tag=’‘, nowait=False)This method cancels a consumer. This does not affect already delivered messages, but it does mean theserver will not send any more messages for that consumer. The client may receive an arbitrary numberof messages in between sending the cancel method and receiving the cancel-ok reply. It may also be sentfrom the server to the client in the event of the consumer being unexpectedly cancelled (i.e. cancelled forany reason other than the server receiving the corresponding basic.cancel from the client). This allowsclients to be notified of the loss of consumers due to events such as queue deletion.
Parameters
• consumer_tag (str) – Identifier for the consumer
• nowait (bool) – Do not expect a Basic.CancelOk response
basic_consume(consumer_callback, queue=’‘, no_ack=False, exclusive=False, con-sumer_tag=None)
Sends the AMQP command Basic.Consume to the broker and binds messages for the consumer_tag to theconsumer callback. If you do not pass in a consumer_tag, one will be automatically generated for you.Returns the consumer tag.
For more information on basic_consume, see: http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.consume
26 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
Parameters
• consumer_callback (method) – The method to callback when consuming
• queue (str or unicode) – The queue to consume from
• no_ack (bool) – Tell the broker to not expect a response
• exclusive (bool) – Don’t allow other consumers on the queue
• consumer_tag (str or unicode) – Specify your own consumer tag
Return type str
basic_get(queue=None, no_ack=False)Get a single message from the AMQP broker. The callback method signature should have 3 parameters:The method frame, header frame and the body, like the consumer callback for Basic.Consume.
Parameters
• queue (str or unicode) – The queue to get a message from
• no_ack (bool) – Tell the broker to not expect a reply
Return type (None, None, None)|(spec.Basic.Get, spec.Basic.Properties, str or unicode)
basic_nack(delivery_tag=None, multiple=False, requeue=True)This method allows a client to reject one or more incoming messages. It can be used to interrupt and cancellarge incoming messages, or return untreatable messages to their original queue.
Parameters
• delivery-tag (int) – The server-assigned delivery tag
• multiple (bool) – If set to True, the delivery tag is treated as “up to and including”, so thatmultiple messages can be acknowledged with a single method. If set to False, the deliverytag refers to a single message. If the multiple field is 1, and the delivery tag is zero, thisindicates acknowledgement of all outstanding messages.
• requeue (bool) – If requeue is true, the server will attempt to requeue the message. Ifrequeue is false or the requeue attempt fails the messages are discarded or dead-lettered.
basic_publish(exchange, routing_key, body, properties=None, mandatory=False, immedi-ate=False)
Publish to the channel with the given exchange, routing key and body. For more information on ba-sic_publish and what the parameters do, see:
http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.publish
Parameters
• exchange (str or unicode) – The exchange to publish to
• routing_key (str or unicode) – The routing key to bind on
• body (str or unicode) – The message body
• properties (pika.spec.Properties) – Basic.properties
• mandatory (bool) – The mandatory flag
• immediate (bool) – The immediate flag
basic_qos(prefetch_size=0, prefetch_count=0, all_channels=False)Specify quality of service. This method requests a specific quality of service. The QoS can be specifiedfor the current channel or for all channels on the connection. The client can request that messages be sent
2.3. Using the BlockingChannel 27
pika Documentation, Release 0.9.9
in advance so that when the client finishes processing a message, the following message is already heldlocally, rather than needing to be sent down the channel. Prefetching gives a performance improvement.
Parameters
• prefetch_size (int) – This field specifies the prefetch window size. The server will senda message in advance if it is equal to or smaller in size than the available prefetch size(and also falls into other prefetch limits). May be set to zero, meaning “no specific limit”,although other prefetch limits may still apply. The prefetch-size is ignored if the no-ackoption is set.
• prefetch_count (int) – Specifies a prefetch window in terms of whole messages. Thisfield may be used in combination with the prefetch-size field; a message will only be sentin advance if both prefetch windows (and those at the channel and connection level) allowit. The prefetch-count is ignored if the no-ack option is set.
• all_channels (bool) – Should the QoS apply to all channels
basic_recover(requeue=False)This method asks the server to redeliver all unacknowledged messages on a specified channel. Zero ormore messages may be redelivered. This method replaces the asynchronous Recover.
Parameters requeue (bool) – If False, the message will be redelivered to the original recipient.If True, the server will attempt to requeue the message, potentially then delivering it to analternative subscriber.
basic_reject(delivery_tag=None, requeue=True)Reject an incoming message. This method allows a client to reject a message. It can be used to interruptand cancel large incoming messages, or return untreatable messages to their original queue.
Parameters
• delivery-tag (int) – The server-assigned delivery tag
• requeue (bool) – If requeue is true, the server will attempt to requeue the message. Ifrequeue is false or the requeue attempt fails the messages are discarded or dead-lettered.
cancel()Cancel the consumption of a queue, rejecting all pending messages. This should only work with thegenerator based BlockingChannel.consume method. If you’re looking to cancel a consumer issues withBlockingChannel.basic_consume then you should call BlockingChannel.basic_cancel.
Return int The number of messages requeued by Basic.Nack
close(reply_code=0, reply_text=’Normal Shutdown’)Will invoke a clean shutdown of the channel with the AMQP Broker.
Parameters
• reply_code (int) – The reply code to close the channel with
• reply_text (str) – The reply text to close the channel with
confirm_delivery(nowait=False)Turn on Confirm mode in the channel.
For more information see: http://www.rabbitmq.com/extensions.html#confirms
Parameters nowait (bool) – Do not send a reply frame (Confirm.SelectOk)
consume(queue)Blocking consumption of a queue instead of via a callback. This method is a generator that returns mes-sages a tuple of method, properties, and body.
28 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
Example:
for method, properties, body in channel.consume(‘queue’): print body chan-nel.basic_ack(method.delivery_tag)
You should call BlockingChannel.cancel() when you escape out of the generator loop. Also note this turnson forced data events to make sure that any acked messages actually get acked.
Parameters queue (str or unicode) – The queue name to consume
Return type tuple(spec.Basic.Deliver, spec.BasicProperties, str or unicode)
consumer_tagsProperty method that returns a list of currently active consumers
Return type list
exchange_bind(destination=None, source=None, routing_key=’‘, nowait=False, arguments=None)Bind an exchange to another exchange.
Parameters
• destination (str or unicode) – The destination exchange to bind
• source (str or unicode) – The source exchange to bind to
• routing_key (str or unicode) – The routing key to bind on
• nowait (bool) – Do not wait for an Exchange.BindOk
• arguments (dict) – Custom key/value pair arguments for the binding
exchange_declare(exchange=None, exchange_type=’direct’, passive=False, durable=False,auto_delete=False, internal=False, nowait=False, arguments=None,type=None)
This method creates an exchange if it does not already exist, and if the exchange exists, verifies that it isof the correct and expected class.
If passive set, the server will reply with Declare-Ok if the exchange already exists with the same name, andraise an error if not and if the exchange does not already exist, the server MUST raise a channel exceptionwith reply code 404 (not found).
Parameters
• exchange (str or unicode) – The exchange name consists of a non-empty sequence ofthese characters: letters, digits, hyphen, underscore, period, or colon.
• exchange_type (str) – The exchange type to use
• passive (bool) – Perform a declare or just check to see if it exists
• durable (bool) – Survive a reboot of RabbitMQ
• auto_delete (bool) – Remove when no more queues are bound to it
• internal (bool) – Can only be published to by other exchanges
• nowait (bool) – Do not expect an Exchange.DeclareOk response
• arguments (dict) – Custom key/value pair arguments for the exchange
• type (str) – The deprecated exchange type parameter
exchange_delete(exchange=None, if_unused=False, nowait=False)Delete the exchange.
Parameters
2.3. Using the BlockingChannel 29
pika Documentation, Release 0.9.9
• exchange (str or unicode) – The exchange name
• if_unused (bool) – only delete if the exchange is unused
• nowait (bool) – Do not wait for an Exchange.DeleteOk
exchange_unbind(destination=None, source=None, routing_key=’‘, nowait=False, argu-ments=None)
Unbind an exchange from another exchange.
Parameters
• destination (str or unicode) – The destination exchange to unbind
• source (str or unicode) – The source exchange to unbind from
• routing_key (str or unicode) – The routing key to unbind
• nowait (bool) – Do not wait for an Exchange.UnbindOk
• arguments (dict) – Custom key/value pair arguments for the binding
flow(callback, active)Turn Channel flow control off and on. Pass a callback to be notified of the response from the server. activeis a bool. Callback should expect a bool in response indicating channel flow state. For more information,please reference:
http://www.rabbitmq.com/amqp-0-9-1-reference.html#channel.flow
Parameters
• callback (method) – The callback method
• active (bool) – Turn flow on or off
force_data_events(enable)Turn on and off forcing the blocking adapter to stop and look to see if there are any frames from RabbitMQin the read buffer. By default the BlockingChannel will check for a read after every RPC command whichcan cause performance to degrade in scenarios where you do not care if RabbitMQ is trying to send RPCcommands to your client connection.
Examples of RPC commands of this sort are:
•Heartbeats
•Connection.Close
•Channel.Close
•Basic.Return
•Basic.Ack and Basic.Nack when using delivery confirmations
Turning off forced data events can be a bad thing and prevents your client from properly communicat-ing with RabbitMQ. Forced data events were added in 0.9.6 to enforce proper channel behavior whencommunicating with RabbitMQ.
Note that the BlockingConnection also has the constant WRITE_TO_READ_RATIO which forces theconnection to stop and try and read after writing the number of frames specified in the constant. This is away to force the client to received these types of frames in a very publish/write IO heavy workload.
Parameters enable (bool) – Set to False to disable
is_closedReturns True if the channel is closed.
Return type bool
30 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
is_closingReturns True if the channel is closing.
Return type bool
is_openReturns True if the channel is open.
Return type bool
open()Open the channel
queue_bind(queue, exchange, routing_key=None, nowait=False, arguments=None)Bind the queue to the specified exchange
Parameters
• queue (str or unicode) – The queue to bind to the exchange
• exchange (str or unicode) – The source exchange to bind to
• routing_key (str or unicode) – The routing key to bind on
• nowait (bool) – Do not wait for a Queue.BindOk
• arguments (dict) – Custom key/value pair arguments for the binding
queue_declare(queue=’‘, passive=False, durable=False, exclusive=False, auto_delete=False,nowait=False, arguments=None)
Declare queue, create if needed. This method creates or checks a queue. When creating a new queue theclient can specify various properties that control the durability of the queue and its contents, and the levelof sharing for the queue.
Leave the queue name empty for a auto-named queue in RabbitMQ
Parameters
• queue (str or unicode) – The queue name
• passive (bool) – Only check to see if the queue exists
• durable (bool) – Survive reboots of the broker
• exclusive (bool) – Only allow access by the current connection
• auto_delete (bool) – Delete after consumer cancels or disconnects
• nowait (bool) – Do not wait for a Queue.DeclareOk
• arguments (dict) – Custom key/value arguments for the queue
queue_delete(queue=’‘, if_unused=False, if_empty=False, nowait=False)Delete a queue from the broker.
Parameters
• queue (str or unicode) – The queue to delete
• if_unused (bool) – only delete if it’s unused
• if_empty (bool) – only delete if the queue is empty
• nowait (bool) – Do not wait for a Queue.DeleteOk
queue_purge(queue=’‘, nowait=False)Purge all of the messages from the specified queue
Parameters
2.3. Using the BlockingChannel 31
pika Documentation, Release 0.9.9
• queue (str or unicode) – The queue to purge
• nowait (bool) – Do not expect a Queue.PurgeOk response
queue_unbind(queue=’‘, exchange=None, routing_key=’‘, arguments=None)Unbind a queue from an exchange.
Parameters
• queue (str or unicode) – The queue to unbind from the exchange
• exchange (str or unicode) – The source exchange to bind from
• routing_key (str or unicode) – The routing key to unbind
• arguments (dict) – Custom key/value pair arguments for the binding
start_consuming()Starts consuming from registered callbacks.
stop_consuming(consumer_tag=None)Sends off the Basic.Cancel to let RabbitMQ know to stop consuming and sets our internal state to exit outof the basic_consume.
tx_commit()Commit a transaction.
tx_rollback()Rollback a transaction.
tx_select()Select standard transaction mode. This method sets the channel to use standard transactions. The clientmust use this method at least once on a channel before using the Commit or Rollback methods.
2.4 Usage Examples
Pika has various methods of use, between the synchronous BlockingConnection adapter and the various asynchronousconnection adapter. The following examples illustrate the various ways that you can use Pika in your projects.
2.4.1 Using URLParameters
Pika has two methods of encapsulating the data that lets it know how to connect to RabbitMQ,pika.connection.ConnectionParameters and pika.connection.URLParameters.
Note: If you’re connecting to RabbitMQ on localhost on port 5672, with the default virtual host of / and the defaultusername and password of guest and guest, you do not need to specify connection parameters when connecting.
Using pika.connection.URLParameters is an easy way to minimize the variables required to connect toRabbitMQ and supports all of the directives that pika.connection.ConnectionParameters supports.
The following is the format for the URLParameters connection value:
scheme://username:password@host:port/virtual_host?key=value&key=value
As you can see, by default, the scheme (amqp, amqps), username, password, host, port and virtual host make up thecore of the URL and any other parameter is passed in as query string values.
32 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
Example Connection URLS
The default connection URL connects to the / virtual host as guest using the guest password on localhost port 5672.Note the forwardslash in the URL is encoded to %2F:
amqp://guest:guest@localhost:5672/%2F
Connect to a host rabbit1 as the user www-data using the password rabbit_pwd on the virtual host web_messages:
amqp://www-data:rabbit_pwd@rabbit1/web_messages
Connecting via SSL is pretty easy too. To connect via SSL for the previous example, simply change the scheme toamqps. If you do not specify a port, Pika will use the default SSL port of 5671:
amqp://www-data:rabbit_pwd@rabbit1/web_messages
If you’re looking to tweak other parameters, such as enabling heartbeats, simply add the key/value pair as a querystring value. The following builds upon the SSL connection, enabling heartbeats every 30 seconds:
amqp://www-data:rabbit_pwd@rabbit1/web_messages?heartbeat_interval=30
Options that are available as query string values:
• backpressure_detection: Pass in a value of t to enable backpressure detection, it is disabled by default.
• channel_max: Alter the default channel maximum by passing in a 32-bit integer value here
• connection_attempts: Alter the default of 1 connection attempt by passing in an integer value here 2.
• frame_max: Alter the default frame maximum size value by passing in a long integer value 3.
• heartbeat_interval: Pass a value greater than zero to enable heartbeats between the server and your application.The integer value you pass here will be the number of seconds between heartbeats.
• locale: Set the locale of the client using underscore delimited posix Locale code in ll_CC format (en_US, pt_BR,de_DE).
• retry_delay: The number of seconds to wait before attempting to reconnect on a failed connection, if connec-tion_attempts is > 0.
• socket_timeout: Change the default socket timeout duration from 0.25 seconds to another integer or float value.Adjust with caution.
• ssl_options: A url encoded dict of values for the SSL connection. The available keys are:
– ca_certs
– cert_reqs
– certfile
– keyfile
– ssl_version
For an information on what the ssl_options can be set to reference the official Python documentation. Here is anexample of setting the client certificate and key:
amqp://www-data:rabbit_pwd@rabbit1/web_messages?heartbeat_interval=30&ssl_options=%7B%27keyfile%27%3A+%27%2Fetc%2Fssl%2Fmykey.pem%27%2C+%27certfile%27%3A+%27%2Fetc%2Fssl%2Fmycert.pem%27%7D
The following example demonstrates how to generate the ssl_options string with Python’s urllib:
2 The pika.adapters.blocking_connection.BlockingConnection adapter does not respect the connection_attempts parame-ter.
3 The AMQP specification states that a server can reject a request for a frame size larger than the value it passes during content negotiation.
2.4. Usage Examples 33
pika Documentation, Release 0.9.9
import urlliburllib.urlencode({’ssl_options’: {’certfile’: ’/etc/ssl/mycert.pem’, ’keyfile’: ’/etc/ssl/mykey.pem’}})
2.4.2 Connecting to RabbitMQ with Callback-Passing Style
When you connect to RabbitMQ with an asynchronous adapter, you are writing event oriented code. The connectionadapter will block on the IOLoop that is watching to see when pika should read data from and write data to RabbitMQ.Because you’re now blocking on the IOLoop, you will receive callback notifications when specific events happen.
Example Code
In the example, there are three steps that take place:
1. Setup the connection to RabbitMQ
2. Start the IOLoop
3. Once connected, the on_open method will be called by Pika with a handle to the connection. In this method, anew channel will be opened on the connection.
4. Once the channel is opened, you can do your other actions, whether they be publishing messages, consumingmessages or other RabbitMQ related activities.:
import pika
# Step #3def on_open(connection):
connection.channel(on_channel_open)
# Step #4def on_channel_open(channel):
channel.basic_publish(’exchange_name’,’routing_key’,’Test Message’,pika.BasicProperties(content_type=’text/plain’,
type=’example’))
# Step #1: Connect to RabbitMQconnection = pika.SelectConnection(on_open_callback=on_open)
try:# Step #2 - Block on the IOLoopconnection.ioloop.start()
# Catch a Keyboard Interrupt to make sure that the connection is closed cleanlyexcept KeyboardInterrupt:
# Gracefully close the connectionconnection.close()
# Start the IOLoop again so Pika can communicate, it will stop on its own when the connection is closedconnection.ioloop.start()
2.4.3 Using the Blocking Connection to get a message from RabbitMQ
The BlockingChannel.basic_get method will return a tuple with the members.
34 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
If the server returns a message, the first item in the tuple will be a pika.spec.Basic.GetOk object with thecurrent message count, the redelivered flag, the routing key that was used to put the message in the queue, and theexchange the message was published to. The second item will be a pika.spec.BasicProperties object andthe third will be the message body.
If the server did not return a message a tuple of None, None, None will be returned.
Example of getting a message and acknowledging it:
import pika
connection = pika.BlockingConnection()channel = connection.channel()method_frame, header_frame, body = channel.basic_get(’test’)if method_frame:
print method_frame, header_frame, bodychannel.basic_ack(method_frame.delivery_tag)
else:print ’No message returned’
2.4.4 Using the Blocking Connection to consume messages from RabbitMQ
The BlockingChannel.basic_consume method assign a callback method to be called every time that Rab-bitMQ delivers messages to your consuming application.
When pika calls your method, it will pass in the channel, a pika.spec.Basic.Deliver object with the deliverytag, the redelivered flag, the routing key that was used to put the message in the queue, and the exchange the messagewas published to. The third argument will be a pika.spec.BasicProperties object and the last will be themessage body.
Example of consuming messages and acknowledging them:
import pika
def on_message(channel, method_frame, header_frame, body):print method_frame.delivery_tagprint bodyprintchannel.basic_ack(delivery_tag=method_frame.delivery_tag)
connection = pika.BlockingConnection()channel = connection.channel()channel.basic_consume(on_message, ’test’)try:
channel.start_consuming()except KeyboardInterrupt:
channel.stop_consuming()connection.close()
2.4.5 Using the BlockingChannel.consume generator to consume messages
The BlockingChannel.consume method is a generator that will return a tuple of method, properties and body.
When you escape out of the loop, be sure to call consumer.cancel() to return any unprocessed messages.
Example of consuming messages and acknowledging them:
2.4. Usage Examples 35
pika Documentation, Release 0.9.9
import pika
connection = pika.BlockingConnection()channel = connection.channel()
# Get ten messages and break outfor method_frame, properties, body in channel.consume(’test’):
# Display the message partsprint method_frameprint propertiesprint body
# Acknowledge the messagechannel.basic_ack(method_frame.delivery_tag)
# Escape out of the loop after 10 messagesif method_frame.delivery_tag == 10:
break
# Cancel the consumer and return any pending messagesrequeued_messages = channel.cancel()print ’Requeued %i messages’ % requeued_messages
# Close the channel and the connectionchannel.close()connection.close()
If you have pending messages in the test queue, your output should look something like:
(pika)gmr-0x02:pika gmr$ python blocking_nack.py<Basic.Deliver([’consumer_tag=ctag1.0’, ’redelivered=True’, ’routing_key=test’, ’delivery_tag=1’, ’exchange=test’])><BasicProperties([’delivery_mode=1’, ’content_type=text/plain’])>Hello World!<Basic.Deliver([’consumer_tag=ctag1.0’, ’redelivered=True’, ’routing_key=test’, ’delivery_tag=2’, ’exchange=test’])><BasicProperties([’delivery_mode=1’, ’content_type=text/plain’])>Hello World!<Basic.Deliver([’consumer_tag=ctag1.0’, ’redelivered=True’, ’routing_key=test’, ’delivery_tag=3’, ’exchange=test’])><BasicProperties([’delivery_mode=1’, ’content_type=text/plain’])>Hello World!<Basic.Deliver([’consumer_tag=ctag1.0’, ’redelivered=True’, ’routing_key=test’, ’delivery_tag=4’, ’exchange=test’])><BasicProperties([’delivery_mode=1’, ’content_type=text/plain’])>Hello World!<Basic.Deliver([’consumer_tag=ctag1.0’, ’redelivered=True’, ’routing_key=test’, ’delivery_tag=5’, ’exchange=test’])><BasicProperties([’delivery_mode=1’, ’content_type=text/plain’])>Hello World!<Basic.Deliver([’consumer_tag=ctag1.0’, ’redelivered=True’, ’routing_key=test’, ’delivery_tag=6’, ’exchange=test’])><BasicProperties([’delivery_mode=1’, ’content_type=text/plain’])>Hello World!<Basic.Deliver([’consumer_tag=ctag1.0’, ’redelivered=True’, ’routing_key=test’, ’delivery_tag=7’, ’exchange=test’])><BasicProperties([’delivery_mode=1’, ’content_type=text/plain’])>Hello World!<Basic.Deliver([’consumer_tag=ctag1.0’, ’redelivered=True’, ’routing_key=test’, ’delivery_tag=8’, ’exchange=test’])><BasicProperties([’delivery_mode=1’, ’content_type=text/plain’])>Hello World!<Basic.Deliver([’consumer_tag=ctag1.0’, ’redelivered=True’, ’routing_key=test’, ’delivery_tag=9’, ’exchange=test’])><BasicProperties([’delivery_mode=1’, ’content_type=text/plain’])>Hello World!
36 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
<Basic.Deliver([’consumer_tag=ctag1.0’, ’redelivered=True’, ’routing_key=test’, ’delivery_tag=10’, ’exchange=test’])><BasicProperties([’delivery_mode=1’, ’content_type=text/plain’])>Hello World!Requeued 1894 messages
2.4.6 Comparing Message Publishing with BlockingConnection and SelectConnec-tion
For those doing simple, non-asynchronous programing, pika.adapters.blocking_connection.BlockingConnection()proves to be the easiest way to get up and running with Pika to publish messages.
In the following example, a connection is made to RabbitMQ listening to port 5672 on localhost using the usernameguest and password guest and virtual host /. Once connected, a channel is opened and a message is published to thetest_exchange exchange using the test_routing_key routing key. The BasicProperties value passed in sets the messageto delivery mode 1 (non-persisted) with a content-type of text/plain. Once the message is published, the connection isclosed:
import pika
parameters = pika.URLParameters(’amqp://guest:guest@localhost:5672/%2F’)
connection = pika.BlockingConnection(paramters)
channel = connection.channel()
channel.basic_publish(’test_exchange’,’test_routing_key’,’message body value’,pika.BasicProperties(content_type=’text/plain’,
delivery_mode=1))
connection.close()
In contrast, using pika.adapters.select_connection.SelectConnection() and the other asyn-chronous adapters is more complicated and less pythonic, but when used with other asynchronous services can havetremendous performance improvements. In the following code example, all of the same parameters and values areused as were used in the previous example:
import pika
# Step #3def on_open(connection):
connection.channel(on_channel_open)
# Step #4def on_channel_open(channel):
channel.basic_publish(’test_exchange’,’test_routing_key’,’message body value’,pika.BasicProperties(content_type=’text/plain’,
delivery_mode=1))
connection.close()
2.4. Usage Examples 37
pika Documentation, Release 0.9.9
# Step #1: Connect to RabbitMQparameters = pika.URLParameters(’amqp://guest:guest@localhost:5672/%2F’)
connection = pika.SelectConnection(parameters=parameters,on_open_callback=on_open)
try:
# Step #2 - Block on the IOLoopconnection.ioloop.start()
# Catch a Keyboard Interrupt to make sure that the connection is closed cleanlyexcept KeyboardInterrupt:
# Gracefully close the connectionconnection.close()
# Start the IOLoop again so Pika can communicate, it will stop on its own when the connection is closedconnection.ioloop.start()
2.4.7 Using Delivery Confirmations with the BlockingConnection
The following code demonstrates how to turn on delivery confirmations with the BlockingConnection and how tocheck for confirmation from RabbitMQ:
import pika
# Open a connection to RabbitMQ on localhost using all default parametersconnection = pika.BlockingConnection()
# Open the channelchannel = connection.channel()
# Declare the queuechannel.queue_declare(queue="test", durable=True, exclusive=False, auto_delete=False)
# Turn on delivery confirmationschannel.confirm_delivery()
# Send a messageif channel.basic_publish(exchange=’test’,
routing_key=’test’,body=’Hello World!’,properties=pika.BasicProperties(content_type=’text/plain’,
delivery_mode=1)):print ’Message publish was confirmed’
else:print ’Message could not be confirmed’
2.4.8 Ensuring message delivery with the mandatory flag
The following example demonstrates how to check if a message is delivered by setting the mandatory flag and checkingthe return result when using the BlockingConnection:
38 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
import pika
# Open a connection to RabbitMQ on localhost using all default parametersconnection = pika.BlockingConnection()
# Open the channelchannel = connection.channel()
# Declare the queuechannel.queue_declare(queue="test", durable=True, exclusive=False, auto_delete=False)
# Send a messageif channel.basic_publish(exchange=’test’,
routing_key=’test’,body=’Hello World!’,properties=pika.BasicProperties(content_type=’text/plain’,
delivery_mode=1),mandatory=True):
print ’Message was published’else:
print ’Message was returned’
2.4.9 Asynchronous consumer example
The following example implements a consumer that will respond to RPC commands sent from RabbitMQ. For ex-ample, it will reconnect if RabbitMQ closes the connection and will shutdown if RabbitMQ cancels the consumer orcloses the channel. While it may look intimidating, each method is very short and represents a individual actions thata consumer can do.
consumer.py:
import loggingimport pika
LOG_FORMAT = (’%(levelname) -10s %(asctime)s %(name) -30s %(funcName) ’’-35s %(lineno) -5d: %(message)s’)
LOGGER = logging.getLogger(__name__)
class ExampleConsumer(object):"""This is an example consumer that will handle unexpected interactionswith RabbitMQ such as channel and connection closures.
If RabbitMQ closes the connection, it will reopen it. You shouldlook at the output, as there are limited reasons why the connection maybe closed, which usually are tied to permission related issues orsocket timeouts.
If the channel is closed, it will indicate a problem with one of thecommands that were issued and that should surface in the output as well.
"""EXCHANGE = ’message’EXCHANGE_TYPE = ’topic’QUEUE = ’text’ROUTING_KEY = ’example.text’
2.4. Usage Examples 39
pika Documentation, Release 0.9.9
def __init__(self, amqp_url):"""Create a new instance of the consumer class, passing in the AMQPURL used to connect to RabbitMQ.
:param str amqp_url: The AMQP url to connect with
"""self._connection = Noneself._channel = Noneself._consumer_tag = Noneself._url = amqp_url
def connect(self):"""This method connects to RabbitMQ, returning the connection handle.When the connection is established, the on_connection_open methodwill be invoked by pika.
:rtype: pika.SelectConnection
"""LOGGER.info(’Connecting to %s’, self._url)return pika.SelectConnection(pika.URLParameters(self._url),
self.on_connection_open)
def close_connection(self):"""This method closes the connection to RabbitMQ."""LOGGER.info(’Closing connection’)self._connection.close()
def add_on_connection_close_callback(self):"""This method adds an on close callback that will be invoked by pikawhen RabbitMQ closes the connection to the publisher unexpectedly.
"""LOGGER.info(’Adding connection close callback’)self._connection.add_on_close_callback(self.on_connection_closed)
def on_connection_closed(self, method_frame):"""This method is invoked by pika when the connection to RabbitMQ isclosed unexpectedly. Since it is unexpected, we will reconnect toRabbitMQ if it disconnects.
:param pika.frame.Method method_frame: The method frame from RabbitMQ
"""LOGGER.warning(’Server closed connection, reopening: (%s) %s’,
method_frame.method.reply_code,method_frame.method.reply_text)
self._channel = Noneself._connection = self.connect()
def on_connection_open(self, unused_connection):"""This method is called by pika once the connection to RabbitMQ hasbeen established. It passes the handle to the connection object incase we need it, but in this case, we’ll just mark it unused.
:type unused_connection: pika.SelectConnection
40 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
"""LOGGER.info(’Connection opened’)self.add_on_connection_close_callback()self.open_channel()
def add_on_channel_close_callback(self):"""This method tells pika to call the on_channel_closed method ifRabbitMQ unexpectedly closes the channel.
"""LOGGER.info(’Adding channel close callback’)self._channel.add_on_close_callback(self.on_channel_closed)
def on_channel_closed(self, method_frame):"""Invoked by pika when RabbitMQ unexpectedly closes the channel.Channels are usually closed if you attempt to do something thatviolates the protocol, such as redeclare an exchange or queue withdifferent paramters. In this case, we’ll close the connectionto shutdown the object.
:param pika.frame.Method method_frame: The Channel.Close method frame
"""LOGGER.warning(’Channel was closed: (%s) %s’,
method_frame.method.reply_code,method_frame.method.reply_text)
self._connection.close()
def on_channel_open(self, channel):"""This method is invoked by pika when the channel has been opened.The channel object is passed in so we can make use of it.
Since the channel is now open, we’ll declare the exchange to use.
:param pika.channel.Channel channel: The channel object
"""LOGGER.info(’Channel opened’)self._channel = channelself.add_on_channel_close_callback()self.setup_exchange(self.EXCHANGE)
def setup_exchange(self, exchange_name):"""Setup the exchange on RabbitMQ by invoking the Exchange.Declare RPCcommand. When it is complete, the on_exchange_declareok method willbe invoked by pika.
:param str|unicode exchange_name: The name of the exchange to declare
"""LOGGER.info(’Declaring exchange %s’, exchange_name)self._channel.exchange_declare(self.on_exchange_declareok,
exchange_name,self.EXCHANGE_TYPE)
def on_exchange_declareok(self, unused_frame):"""Invoked by pika when RabbitMQ has finished the Exchange.Declare RPCcommand.
2.4. Usage Examples 41
pika Documentation, Release 0.9.9
:param pika.Frame.Method unused_frame: Exchange.DeclareOk response frame
"""LOGGER.info(’Exchange declared’)self.setup_queue(self.QUEUE)
def setup_queue(self, queue_name):"""Setup the queue on RabbitMQ by invoking the Queue.Declare RPCcommand. When it is complete, the on_queue_declareok method willbe invoked by pika.
:param str|unicode queue_name: The name of the queue to declare.
"""LOGGER.info(’Declaring queue %s’, queue_name)self._channel.queue_declare(self.on_queue_declareok, queue_name)
def on_queue_declareok(self, method_frame):"""Method invoked by pika when the Queue.Declare RPC call made insetup_queue has completed. In this method we will bind the queueand exchange together with the routing key by issuing the Queue.BindRPC command. When this command is complete, the on_bindok method willbe invoked by pika.
:param pika.frame.Method method_frame: The Queue.DeclareOk frame
"""LOGGER.info(’Binding %s to %s with %s’,
self.EXCHANGE, self.QUEUE, self.ROUTING_KEY)self._channel.queue_bind(self.on_bindok, self.QUEUE,
self.EXCHANGE, self.ROUTING_KEY)
def add_on_cancel_callback(self):"""Add a callback that will be invoked if RabbitMQ cancels the consumerfor some reason. If RabbitMQ does cancel the consumer,on_consumer_cancelled will be invoked by pika.
"""LOGGER.info(’Adding consumer cancellation callback’)self._channel.add_on_cancel_callback(self.on_consumer_cancelled)
def on_consumer_cancelled(self, method_frame):"""Invoked by pika when RabbitMQ sends a Basic.Cancel for a consumerreceiving messages.
:param pika.frame.Method method_frame: The Basic.Cancel frame
"""LOGGER.info(’Consumer was cancelled remotely, shutting down: %r’,
method_frame)self._channel.close()
def acknowledge_message(self, delivery_tag):"""Acknowledge the message delivery from RabbitMQ by sending aBasic.Ack RPC method for the delivery tag.
:param int delivery_tag: The delivery tag from the Basic.Deliver frame
42 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
"""LOGGER.info(’Acknowledging message %s’, delivery_tag)self._channel.basic_ack(delivery_tag)
def on_message(self, unused_channel, basic_deliver, properties, body):"""Invoked by pika when a message is delivered from RabbitMQ. Thechannel is passed for your convenience. The basic_deliver object thatis passed in carries the exchange, routing key, delivery tag anda redelivered flag for the message. The properties passed in is aninstance of BasicProperties with the message properties and the bodyis the message that was sent.
:param pika.channel.Channel unused_channel: The channel object:param pika.Spec.Basic.Deliver: basic_deliver method:param pika.Spec.BasicProperties: properties:param str|unicode body: The message body
"""LOGGER.info(’Received message # %s from %s: %s’,
basic_deliver.delivery_tag, properties.app_id, body)self.acknowledge_message(basic_deliver.delivery_tag)
def on_cancelok(self, unused_frame):"""This method is invoked by pika when RabbitMQ acknowledges thecancellation of a consumer. At this point we will close the connectionwhich will automatically close the channel if it’s open.
:param pika.frame.Method unused_frame: The Basic.CancelOk frame
"""LOGGER.info(’RabbitMQ acknowledged the cancellation of the consumer’)self.close_connection()
def stop_consuming(self):"""Tell RabbitMQ that you would like to stop consuming by sending theBasic.Cancel RPC command.
"""LOGGER.info(’Sending a Basic.Cancel RPC command to RabbitMQ’)self._channel.basic_cancel(self.on_cancelok, self._consumer_tag)
def start_consuming(self):"""This method sets up the consumer by first callingadd_on_cancel_callback so that the object is notified if RabbitMQcancels the consumer. It then issues the Basic.Consume RPC commandwhich returns the consumer tag that is used to uniquely identify theconsumer with RabbitMQ. We keep the value to use it when we want tocancel consuming. The on_message method is passed in as a callback pikawill invoke when a message is fully received.
"""LOGGER.info(’Issuing consumer related RPC commands’)self.add_on_cancel_callback()self._consumer_tag = self._channel.basic_consume(self.on_message,
self.QUEUE)
def on_bindok(self, unused_frame):"""Invoked by pika when the Queue.Bind method has completed. At this
2.4. Usage Examples 43
pika Documentation, Release 0.9.9
point we will start consuming messages by calling start_consumingwhich will invoke the needed RPC commands to start the process.
:param pika.frame.Method unused_frame: The Queue.BindOk response frame
"""LOGGER.info(’Queue bound’)self.start_consuming()
def close_channel(self):"""Call to close the channel with RabbitMQ cleanly by issuing theChannel.Close RPC command.
"""LOGGER.info(’Closing the channel’)self._channel.close()
def open_channel(self):"""Open a new channel with RabbitMQ by issuing the Channel.Open RPCcommand. When RabbitMQ responds that the channel is open, theon_channel_open callback will be invoked by pika.
"""LOGGER.info(’Creating a new channel’)self._connection.channel(on_open_callback=self.on_channel_open)
def run(self):"""Run the example consumer by connecting to RabbitMQ and thenstarting the IOLoop to block and allow the SelectConnection to operate.
"""self._connection = self.connect()self._connection.ioloop.start()
def stop(self):"""Cleanly shutdown the connection to RabbitMQ by stopping the consumerwith RabbitMQ. When RabbitMQ confirms the cancellation, on_cancelokwill be invoked by pika, which will then closing the channel andconnection. The IOLoop is started again because this method is invokedwhen CTRL-C is pressed raising a KeyboardInterrupt exception. Thisexception stops the IOLoop which needs to be running for pika tocommunicate with RabbitMQ. All of the commands issued prior to startingthe IOLoop will be buffered but not processed.
"""LOGGER.info(’Stopping’)self.stop_consuming()self._connection.ioloop.start()
def main():logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)example = ExampleConsumer(’amqp://guest:guest@localhost:5672/%2F’)try:
example.run()except KeyboardInterrupt:
example.stop()
44 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
if __name__ == ’__main__’:main()
2.4.10 Asynchronous publisher example
The following example implements a publisher that will respond to RPC commands sent from RabbitMQ and usesdelivery confirmations. It will reconnect if RabbitMQ closes the connection and will shutdown if RabbitMQ closes thechannel. While it may look intimidating, each method is very short and represents a individual actions that a publishercan do.
publisher.py:
import loggingimport pikaimport time
LOG_FORMAT = (’%(levelname) -10s %(asctime)s %(name) -30s %(funcName) ’’-35s %(lineno) -5d: %(message)s’)
LOGGER = logging.getLogger(__name__)
class ExamplePublisher(object):"""This is an example publisher that will handle unexpected interactionswith RabbitMQ such as channel and connection closures.
If RabbitMQ closes the connection, it will reopen it. You shouldlook at the output, as there are limited reasons why the connection maybe closed, which usually are tied to permission related issues orsocket timeouts.
It uses delivery confirmations and illustrates one way to keep track ofmessages that have been sent and if they’ve been confirmed by RabbitMQ.
"""EXCHANGE = ’message’EXCHANGE_TYPE = ’topic’PUBLISH_INTERVAL = 1QUEUE = ’text’ROUTING_KEY = ’example.text’
def __init__(self, amqp_url):"""Setup the example publisher object, passing in the URL we will useto connect to RabbitMQ.
:param str amqp_url: The URL for connecting to RabbitMQ
"""self._connection = Noneself._channel = Noneself._deliveries = []self._acked = 0self._nacked = 0self._message_number = 0self._stopping = Falseself._url = amqp_url
2.4. Usage Examples 45
pika Documentation, Release 0.9.9
def connect(self):"""This method connects to RabbitMQ, returning the connection handle.When the connection is established, the on_connection_open methodwill be invoked by pika.
:rtype: pika.SelectConnection
"""LOGGER.info(’Connecting to %s’, self._url)return pika.SelectConnection(pika.URLParameters(self._url),
self.on_connection_open)
def close_connection(self):"""This method closes the connection to RabbitMQ."""LOGGER.info(’Closing connection’)self._connection.close()
def add_on_connection_close_callback(self):"""This method adds an on close callback that will be invoked by pikawhen RabbitMQ closes the connection to the publisher unexpectedly.
"""LOGGER.info(’Adding connection close callback’)self._connection.add_on_close_callback(self.on_connection_closed)
def on_connection_closed(self, method_frame):"""This method is invoked by pika when the connection to RabbitMQ isclosed unexpectedly. Since it is unexpected, we will reconnect toRabbitMQ if it disconnects.
:param pika.frame.Method method_frame: The method frame from RabbitMQ
"""LOGGER.warning(’Server closed connection, reopening: (%s) %s’,
method_frame.method.reply_code,method_frame.method.reply_text)
self._channel = Noneself._connection = self.connect()
def on_connection_open(self, unused_connection):"""This method is called by pika once the connection to RabbitMQ hasbeen established. It passes the handle to the connection object incase we need it, but in this case, we’ll just mark it unused.
:type unused_connection: pika.SelectConnection
"""LOGGER.info(’Connection opened’)self.add_on_connection_close_callback()self.open_channel()
def add_on_channel_close_callback(self):"""This method tells pika to call the on_channel_closed method ifRabbitMQ unexpectedly closes the channel.
"""LOGGER.info(’Adding channel close callback’)self._channel.add_on_close_callback(self.on_channel_closed)
46 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
def on_channel_closed(self, method_frame):"""Invoked by pika when RabbitMQ unexpectedly closes the channel.Channels are usually closed if you attempt to do something thatviolates the protocol, such as redeclare an exchange or queue withdifferent paramters. In this case, we’ll close the connectionto shutdown the object.
:param pika.frame.Method method_frame: The Channel.Close method frame
"""LOGGER.warning(’Channel was closed: (%s) %s’,
method_frame.method.reply_code,method_frame.method.reply_text)
self._connection.close()
def on_channel_open(self, channel):"""This method is invoked by pika when the channel has been opened.The channel object is passed in so we can make use of it.
Since the channel is now open, we’ll declare the exchange to use.
:param pika.channel.Channel channel: The channel object
"""LOGGER.info(’Channel opened’)self._channel = channelself.add_on_channel_close_callback()self.setup_exchange(self.EXCHANGE)
def setup_exchange(self, exchange_name):"""Setup the exchange on RabbitMQ by invoking the Exchange.Declare RPCcommand. When it is complete, the on_exchange_declareok method willbe invoked by pika.
:param str|unicode exchange_name: The name of the exchange to declare
"""LOGGER.info(’Declaring exchange %s’, exchange_name)self._channel.exchange_declare(self.on_exchange_declareok,
exchange_name,self.EXCHANGE_TYPE)
def on_exchange_declareok(self, unused_frame):"""Invoked by pika when RabbitMQ has finished the Exchange.Declare RPCcommand.
:param pika.Frame.Method unused_frame: Exchange.DeclareOk response frame
"""LOGGER.info(’Exchange declared’)self.setup_queue(self.QUEUE)
def setup_queue(self, queue_name):"""Setup the queue on RabbitMQ by invoking the Queue.Declare RPCcommand. When it is complete, the on_queue_declareok method willbe invoked by pika.
:param str|unicode queue_name: The name of the queue to declare.
2.4. Usage Examples 47
pika Documentation, Release 0.9.9
"""LOGGER.info(’Declaring queue %s’, queue_name)self._channel.queue_declare(self.on_queue_declareok, queue_name)
def on_queue_declareok(self, method_frame):"""Method invoked by pika when the Queue.Declare RPC call made insetup_queue has completed. In this method we will bind the queueand exchange together with the routing key by issuing the Queue.BindRPC command. When this command is complete, the on_bindok method willbe invoked by pika.
:param pika.frame.Method method_frame: The Queue.DeclareOk frame
"""LOGGER.info(’Binding %s to %s with %s’,
self.EXCHANGE, self.QUEUE, self.ROUTING_KEY)self._channel.queue_bind(self.on_bindok, self.QUEUE,
self.EXCHANGE, self.ROUTING_KEY)
def on_delivery_confirmation(self, method_frame):"""Invoked by pika when RabbitMQ responds to a Basic.Publish RPCcommand, passing in either a Basic.Ack or Basic.Nack frame withthe delivery tag of the message that was published. The delivery tagis an integer counter indicating the message number that was senton the channel via Basic.Publish. Here we’re just doing house keepingto keep track of stats and remove message numbers that we expecta delivery confirmation of from the list used to keep track of messagesthat are pending confirmation.
:param pika.frame.Method method_frame: Basic.Ack or Basic.Nack frame
"""confirmation_type = method_frame.method.NAME.split(’.’)[1].lower()LOGGER.info(’Received %s for delivery tag: %i’,
confirmation_type,method_frame.method.delivery_tag)
if confirmation_type == ’ack’:self._acked += 1
elif confirmation_type == ’nack’:self._nacked += 1
self._deliveries.remove(method_frame.method.delivery_tag)LOGGER.info(’Published %i messages, %i have yet to be confirmed, ’
’%i were acked and %i were nacked’,self._message_number, len(self._deliveries),self._acked, self._nacked)
def enable_delivery_confirmations(self):"""Send the Confirm.Select RPC method to RabbitMQ to enable deliveryconfirmations on the channel. The only way to turn this off is to closethe channel and create a new one.
When the message is confirmed from RabbitMQ, theon_delivery_confirmation method will be invoked passing in a Basic.Ackor Basic.Nack method from RabbitMQ that will indicate which messages itis confirming or rejecting.
"""LOGGER.info(’Issuing Confirm.Select RPC command’)
48 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
self._channel.confirm_delivery(self.on_delivery_confirmation)
def publish_message(self):"""If the class is not stopping, publish a message to RabbitMQ,appending a list of deliveries with the message number that was sent.This list will be used to check for delivery confirmations in theon_delivery_confirmations method.
Once the message has been sent, schedule another message to be sent.The main reason I put scheduling in was just so you can get a good ideaof how the process is flowing by slowing down and speeding up thedelivery intervals by changing the PUBLISH_INTERVAL constant in theclass.
"""if self._stopping:
return
message = ’The current epoch value is %i’ % time.time()properties = pika.BasicProperties(app_id=’example-publisher’,
content_type=’text/plain’)
self._channel.basic_publish(self.EXCHANGE, self.ROUTING_KEY,message, properties)
self._message_number += 1self._deliveries.append(self._message_number)LOGGER.info(’Published message # %i’, self._message_number)self.schedule_next_message()
def schedule_next_message(self):"""If we are not closing our connection to RabbitMQ, schedule anothermessage to be delivered in PUBLISH_INTERVAL seconds.
"""if self._stopping:
returnLOGGER.info(’Scheduling next message for %0.1f seconds’,
self.PUBLISH_INTERVAL)self._connection.add_timeout(self.PUBLISH_INTERVAL,
self.publish_message)
def start_publishing(self):"""This method will enable delivery confirmations and schedule thefirst message to be sent to RabbitMQ
"""LOGGER.info(’Issuing consumer related RPC commands’)self.enable_delivery_confirmations()self.schedule_next_message()
def on_bindok(self, unused_frame):"""This method is invoked by pika when it receives the Queue.BindOkresponse from RabbitMQ. Since we know we’re now setup and bound, it’stime to start publishing."""LOGGER.info(’Queue bound’)self.start_publishing()
def close_channel(self):
2.4. Usage Examples 49
pika Documentation, Release 0.9.9
"""Invoke this command to close the channel with RabbitMQ by sendingthe Channel.Close RPC command.
"""LOGGER.info(’Closing the channel’)self._channel.close()
def open_channel(self):"""This method will open a new channel with RabbitMQ by issuing theChannel.Open RPC command. When RabbitMQ confirms the channel is openby sending the Channel.OpenOK RPC reply, the on_channel_open methodwill be invoked.
"""LOGGER.info(’Creating a new channel’)self._connection.channel(on_open_callback=self.on_channel_open)
def run(self):"""Run the example code by connecting and then starting the IOLoop.
"""self._connection = self.connect()self._connection.ioloop.start()
def stop(self):"""Stop the example by closing the channel and connection. Weset a flag here so that we stop scheduling new messages to bepublished. The IOLoop is started because this method isinvoked by the Try/Catch below when KeyboardInterrupt is caught.Starting the IOLoop again will allow the publisher to cleanlydisconnect from RabbitMQ.
"""LOGGER.info(’Stopping’)self._stopping = Trueself.close_channel()self.close_connection()self._connection.ioloop.start()
def main():logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
# Connect to localhost:5672 as guest with the password guest and virtual host "/" (%2F)example = ExamplePublisher(’amqp://guest:guest@localhost:5672/%2F’)try:
example.run()except KeyboardInterrupt:
example.stop()
if __name__ == ’__main__’:main()
2.5 Frequently Asked Questions
• Is Pika thread safe?
50 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
Pika does not have any notion of threading in the code. If you want to use Pika with threading, makesure you have a Pika connection per thread, created in that thread. It is not safe to share one Pikaconnection across threads.
• How do I report a bug with Pika?
The main Pika repository is hosted on Github and we use the Issue tracker athttps://github.com/pika/pika/issues.
• Is there a mailing list for Pika?
Yes, Pika’s mailing list is available on Google Groups and the email address is [email protected], though traditionally questions about Pika have been asked on theRabbitMQ-Discuss mailing list.
• Is there an IRC channel for Pika?
People knowledgeable about Pika tend to hang out in #pika and #RabbitMQ on irc.freenode.net.
• What versions of Python are supported?
Main developent is currently on the Python 2.6 branch. For a release, Pika passes its tests on thelatest versions of 2.5, 2.6 and 2.7.
• Does Pika work with Python 3?
Not yet.
• How can I contribute to Pika?
You can fork the project on Github and issue Pull Requests when you believe you have somethingsolid to be added to the main repository.
2.6 Version History
2.6.1 0.9.9 - 2013-01-29
Bugfixes
• Only remove the tornado_connection.TornadoConnection file descriptor from the IOLoop if it’s still open (Issue#221)
• Allow messages with no body (Issue #227)
• Allow for empty routing keys (Issue #224)
• Don’t raise an exception when trying to send a frame to a closed connection (Issue #229)
• Only send a Connection.CloseOk if the connection is still open. (Issue #236 - Fix by noleaf)
• Fix timeout threshold in blocking connection - (Issue #232 - Fix by Adam Flynn)
• Fix closing connection while a channel is still open (Issue #230 - Fix by Adam Flynn)
• Fixed misleading warning and exception messages in BaseConnection (Issue #237 - Fix by Tristan Penman)
• Pluralised and altered the wording of the AMQPConnectionError exception (Issue #237 - Fix by Tristan Pen-man)
• Fixed _adapter_disconnect in TornadoConnection class (Issue #237 - Fix by Tristan Penman)
• Fixing hang when closing connection without any channel in BlockingConnection (Issue #244 - Fix by AlesTeska)
2.6. Version History 51
pika Documentation, Release 0.9.9
• Remove the process_timeouts() call in SelectConnection (Issue #239)
• Change the string validation to basestring for host connection parameters (Issue #231)
• Add a poller to the BlockingConnection to address latency issues introduced in Pika 0.9.8 (Issue #242)
• reply_code and reply_text is not set in ChannelException (Issue #250)
• Add the missing constraint parameter for Channel._on_return callback processing (Issue #257 - Fix by patcpsc)
• Channel callbacks not being removed from callback manager when channel is closed or deleted (Issue #261)
2.6.2 0.9.8 - 2012-11-18
Bugfixes
• Channel.queue_declare/BlockingChannel.queue_declare not setting up callbacks property for empty queuename (Issue #218)
• Channel.queue_bind/BlockingChannel.queue_bind not allowing empty routing key
• Connection._on_connection_closed calling wrong method in Channel (Issue #219)
• Fix tx_commit and tx_rollback bugs in BlockingChannel (Issue #217)
2.6.3 0.9.7 - 2012-11-11
New features
• generator based consumer in BlockingChannel (See Using the BlockingChannel.consume generator to consumemessages for example)
Changes
• BlockingChannel._send_method will only wait if explicitly told to
Bugfixes
• Added the exchange “type” parameter back but issue a DeprecationWarning
• Dont require a queue name in Channel.queue_declare()
• Fixed KeyError when processing timeouts (Issue # 215 - Fix by Raphael De Giusti)
• Don’t try and close channels when the connection is closed (Issue #216 - Fix by Charles Law)
• Dont raise UnexpectedFrame exceptions, log them instead
• Handle multiple synchronous RPC calls made without waiting for the call result (Issues #192, #204, #211)
• Typo in docs (Issue #207 Fix by Luca Wehrstedt)
• Only sleep on connection failure when retry attempts are > 0 (Issue #200)
• Bypass _rpc method and just send frames for Basic.Ack, Basic.Nack, Basic.Reject (Issue #205)
2.6.4 0.9.6 - 2012-10-29
New features
• URLParameters
• BlockingChannel.start_consuming() and BlockingChannel.stop_consuming()
52 Chapter 2. Using Pika
pika Documentation, Release 0.9.9
• Delivery Confirmations
• Improved unittests
Major bugfix areas
• Connection handling
• Blocking functionality in the BlockingConnection
• SSL
• UTF-8 Handling
Removals
• pika.reconnection_strategies
• pika.channel.ChannelTransport
• pika.log
• pika.template
• examples directory
2.6.5 0.9.5 - 2011-03-29
Changelog
• Scope changes with adapter IOLoops and CallbackManager allowing for cleaner, multi-threaded operation
• Add support for Confirm.Select with channel.Channel.confirm_delivery()
• Add examples of delivery confirmation to examples (demo_send_confirmed.py)
• Update uses of log.warn with warning.warn for TCP Back-pressure alerting
• License boilerplate updated to simplify license text in source files
• Increment the timeout in select_connection.SelectPoller reducing CPU utilization
• Bug fix in Heartbeat frame delivery addressing issue #35
• Remove abuse of pika.log.method_call through a majority of the code
• Rename of key modules: table to data, frames to frame
• Cleanup of frame module and related classes
• Restructure of tests and test runner
• Update functional tests to respect RABBITMQ_HOST, RABBITMQ_PORT environment variables
• Bug fixes to reconnection_strategies module
• Fix the scale of timeout for PollPoller to be specified in milliseconds
• Remove mutable default arguments in RPC calls
• Add data type validation to RPC calls
• Move optional credentials erasing out of connection.Connection into credentials module
• Add support to allow for additional external credential types
• Add a NullHandler to prevent the ‘No handlers could be found for logger “pika”’ error message when not usingpika.log in a client app at all.
2.6. Version History 53
pika Documentation, Release 0.9.9
• Clean up all examples to make them easier to read and use
• Move documentation into its own repository https://github.com/pika/documentation
• channel.py
– Move channel.MAX_CHANNELS constant from connection.CHANNEL_MAX
– Add default value of None to ChannelTransport.rpc
– Validate callback and acceptable replies parameters in ChannelTransport.RPC
– Remove unused connection attribute from Channel
• connection.py
– Remove unused import of struct
– Remove direct import of pika.credentials.PlainCredentials - Change to import pika.credentials
– Move CHANNEL_MAX to channel.MAX_CHANNELS
– Change ConnectionParameters initialization parameter heartbeat to boolean
– Validate all inbound parameter types in ConnectionParameters
– Remove the Connection._erase_credentials stub method in favor of letting the Credentials object deal withthat itself.
– Warn if the credentials object intends on erasing the credentials and a reconnection strategy other thanNullReconnectionStrategy is specified.
– Change the default types for callback and acceptable_replies in Connection._rpc
– Validate the callback and acceptable_replies data types in Connection._rpc
• adapters.blocking_connection.BlockingConnection
– Addition of _adapter_disconnect to blocking_connection.BlockingConnection
– Add timeout methods to BlockingConnection addressing issue #41
– BlockingConnection didn’t allow you register more than one consumer callback because basic_consumewas overridden to block immediately. New behavior allows you to do so.
– Removed overriding of base basic_consume and basic_cancel methods. Now uses underlying Channelversions of those methods.
– Added start_consuming() method to BlockingChannel to start the consumption loop.
– Updated stop_consuming() to iterate through all the registered consumers in self._consumers and issue abasic_cancel.
54 Chapter 2. Using Pika
CHAPTER
THREE
0.9.9 - 2013-01-29
Bugfixes
• Only remove the tornado_connection.TornadoConnection file descriptor from the IOLoop if it’s still open (Issue#221)
• Allow messages with no body (Issue #227)
• Allow for empty routing keys (Issue #224)
• Don’t raise an exception when trying to send a frame to a closed connection (Issue #229)
• Only send a Connection.CloseOk if the connection is still open. (Issue #236 - Fix by noleaf)
• Fix timeout threshold in blocking connection - (Issue #232 - Fix by Adam Flynn)
• Fix closing connection while a channel is still open (Issue #230 - Fix by Adam Flynn)
• Fixed misleading warning and exception messages in BaseConnection (Issue #237 - Fix by Tristan Penman)
• Pluralised and altered the wording of the AMQPConnectionError exception (Issue #237 - Fix by Tristan Pen-man)
• Fixed _adapter_disconnect in TornadoConnection class (Issue #237 - Fix by Tristan Penman)
• Fixing hang when closing connection without any channel in BlockingConnection (Issue #244 - Fix by AlesTeska)
• Remove the process_timeouts() call in SelectConnection (Issue #239)
• Change the string validation to basestring for host connection parameters (Issue #231)
• Add a poller to the BlockingConnection to address latency issues introduced in Pika 0.9.8 (Issue #242)
• reply_code and reply_text is not set in ChannelException (Issue #250)
• Add the missing constraint parameter for Channel._on_return callback processing (Issue #257 - Fix by patcpsc)
• Channel callbacks not being removed from callback manager when channel is closed or deleted (Issue #261)
55
pika Documentation, Release 0.9.9
56 Chapter 3. 0.9.9 - 2013-01-29
CHAPTER
FOUR
PIKA CORE MODULES AND CLASSES
Note: The following documentation is for Pika development and is not intended to be end-user documentation.
4.1 adapters
Note: The following class level documentation is not intended for use by those using Pika in their applications. Thisdocumentation is for those who are extending Pika or otherwise working on the driver itself. For an overview of howto use adapters, please reference the Connecting to RabbitMQ documentation.
4.1.1 base_connection
Base class extended by connection adapters. This extends the connection.Connection class to encapsulate connectionbehavior but still isolate socket and low level communication.
BaseConnection
class pika.adapters.base_connection.BaseConnection(parameters=None,on_open_callback=None,stop_ioloop_on_close=True)
BaseConnection class that should be extended by connection adapters
BaseConnection.READ = 1
BaseConnection.WRITE = 4
BaseConnection.ERROR = 8
BaseConnection.ERRORS_TO_IGNORE = [11, 11, 4]
BaseConnection.DO_HANDSHAKE = True
BaseConnection.WARN_ABOUT_IOLOOP = False
BaseConnection.add_timeout(deadline, callback_method)Add the callback_method to the IOLoop timer to fire after deadline seconds. Returns a handle to thetimeout
Parameters
57
pika Documentation, Release 0.9.9
• deadline (int) – The number of seconds to wait to call callback
• callback_method (method) – The callback method
Return type str
BaseConnection.close(reply_code=200, reply_text=’Normal shutdown’)Disconnect from RabbitMQ. If there are any open channels, it will attempt to close them prior to fullydisconnecting. Channels which have active consumers will attempt to send a Basic.Cancel to RabbitMQto cleanly stop the delivery of messages prior to closing the channel.
Parameters
• reply_code (int) – The code number for the close
• reply_text (str) – The text reason for the close
BaseConnection.remove_timeout(timeout_id)Remove the timeout from the IOLoop by the ID returned from add_timeout.
Return type str
BaseConnection._adapter_connect()Connect to the RabbitMQ broker
BaseConnection._adapter_disconnect()Invoked if the connection is being told to disconnect
BaseConnection._check_state_on_disconnect()Checks to see if we were in opening a connection with RabbitMQ when we were disconnected and raisesexceptions for the anticipated exception types.
BaseConnection._create_and_connect_to_socket()Create socket and connect to it, using SSL if enabled.
BaseConnection._do_ssl_handshake()Perform SSL handshaking, copied from python stdlib test_ssl.py.
BaseConnection._get_error_code(error_value)Get the error code from the error_value accounting for Python version differences.
Return type int
BaseConnection._flush_outbound()Call the state manager who will figure out that we need to write.
BaseConnection._handle_disconnect()Called internally when the socket is disconnected already
BaseConnection._handle_ioloop_stop()Invoked when the connection is closed to determine if the IOLoop should be stopped or not.
BaseConnection._handle_error(error_value)Internal error handling method. Here we expect a socket.error coming in and will handle different socketerrors differently.
Parameters error_value (int|object) – The inbound error
BaseConnection._handle_events(fd, events, error=None, write_only=False)Handle IO/Event loop events, processing them.
Parameters
• fd (int) – The file descriptor for the events
• events (int) – Events from the IO/Event loop
58 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
• error (int) – Was an error specified
• write_only (bool) – Only handle write events
BaseConnection._handle_read()Read from the socket and call our on_data_available with the data.
BaseConnection._handle_write()Handle any outbound buffer writes that need to take place.
BaseConnection._init_connection_state()Initialize or reset all of our internal state variables for a given connection. If we disconnect and reconnect,all of our state needs to be wiped.
BaseConnection._manage_event_state()Manage the bitmask for reading/writing/error which is used by the io/event handler to specify when thereis an event such as a read or write.
BaseConnection._wrap_socket(sock)Wrap the socket for connecting over SSL.
Return type ssl.SSLSocket
4.1.2 asyncore_connection
Use pika with the stdlib asyncore module
AsyncoreConnection
class pika.adapters.asyncore_connection.AsyncoreConnection(parameters=None,on_open_callback=None,stop_ioloop_on_close=True)
AsyncoreConnection._adapter_connect()Connect to our RabbitMQ broker using AsyncoreDispatcher, then setting Pika’s suggested buffer size forsocket reading and writing. We pass the handle to self so that the AsyncoreDispatcher object can call backinto our various state methods.
PikaDispatcher
class pika.adapters.asyncore_connection.PikaDispatcher(sock=None, map=None,event_callback=None)
PikaDispatcher.READ = 1
PikaDispatcher.WRITE = 4
PikaDispatcher.ERROR = 8
PikaDispatcher.add_timeout(deadline, handler)Add a timeout with with given deadline, should return a timeout id.
Parameters
• deadline (int) – The number of seconds to wait until calling handler
• handler (method) – The method to call at deadline
Return type str
4.1. adapters 59
pika Documentation, Release 0.9.9
PikaDispatcher.readable()
PikaDispatcher.writable()
PikaDispatcher.handle_read()
PikaDispatcher.handle_write()
PikaDispatcher.process_timeouts()Process the self._timeouts event stack
PikaDispatcher.remove_timeout(timeout_id)Remove a timeout if it’s still in the timeout stack
Parameters timeout_id (str) – The timeout id to remove
PikaDispatcher.start()
PikaDispatcher.stop()
PikaDispatcher.update_handler(fileno_unused, events)Set the events to the current events
Parameters
• fileno_unused (int) – The file descriptor
• events (int) – The event mask
4.1.3 blocking_connection
Implement a blocking, procedural style connection adapter on top of the asynchronous core.
BlockingConnection
class pika.adapters.blocking_connection.BlockingConnection(parameters=None,on_open_callback=None,stop_ioloop_on_close=True)
The BlockingConnection adapter is meant for simple implementations where you want to have blocking behav-ior. The behavior layered on top of the async library. Because of the nature of AMQP there are a few callbacksone needs to do, even in a blocking implementation. These include receiving messages from Basic.Deliver,Basic.GetOk, and Basic.Return.
BlockingConnection.WRITE_TO_READ_RATIO = 1000
BlockingConnection.DO_HANDSHAKE = True
BlockingConnection.SLEEP_DURATION = 0.1
BlockingConnection.SOCKET_CONNECT_TIMEOUT = 0.25
BlockingConnection.SOCKET_TIMEOUT_THRESHOLD = 12
BlockingConnection.SOCKET_TIMEOUT_CLOSE_THRESHOLD = 3
BlockingConnection.SOCKET_TIMEOUT_MESSAGE = ‘Timeout exceeded, disconnected’
BlockingConnection.add_timeout(deadline, callback)Add the callback to the IOLoop timer to fire after deadline seconds.
Parameters
• deadline (int) – The number of seconds to wait to call callback
60 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
• callback (method) – The callback method
Return type str
BlockingConnection.channel(channel_number=None)Create a new channel with the next available or specified channel #.
Parameters channel_number (int) – Specify the channel number
BlockingConnection.close(reply_code=200, reply_text=’Normal shutdown’)Disconnect from RabbitMQ. If there are any open channels, it will attempt to close them prior to fullydisconnecting. Channels which have active consumers will attempt to send a Basic.Cancel to RabbitMQto cleanly stop the delivery of messages prior to closing the channel.
Parameters
• reply_code (int) – The code number for the close
• reply_text (str) – The text reason for the close
BlockingConnection.disconnect()Disconnect from the socket
BlockingConnection.process_data_events()Will make sure that data events are processed. Your app can block on this method.
BlockingConnection.process_timeouts()Process the self._timeouts event stack
BlockingConnection.remove_timeout(timeout_id)Remove the timeout from the IOLoop by the ID returned from add_timeout.
Parameters timeout_id (str) – The id of the timeout to remove
BlockingConnection.send_method(channel_number, method_frame, content=None)Constructs a RPC method frame and then sends it to the broker.
Parameters
• channel_number (int) – The channel number for the frame
• method_frame (pika.object.Method) – The method frame to send
• content (tuple) – If set, is a content frame, is tuple of properties and body.
BlockingConnection.sleep(duration)A safer way to sleep than calling time.sleep() directly which will keep the adapter from ignoring framessent from RabbitMQ. The connection will “sleep” or block the number of seconds specified in duration insmall intervals.
Parameters duration (int) – The time to sleep
BlockingConnection._adapter_connect()Connect to the RabbitMQ broker
BlockingConnection._adapter_disconnect()Called if the connection is being requested to disconnect.
BlockingConnection._call_timeout_method(timeout_value)Execute the method that was scheduled to be called.
Parameters timeout_value (dict) – The configuration for the timeout
BlockingConnection._deadline_passed(timeout_id)Returns True if the deadline has passed for the specified timeout_id.
4.1. adapters 61
pika Documentation, Release 0.9.9
Parameters timeout_id (str) – The id of the timeout to check
Return type bool
BlockingConnection._handle_disconnect()Called internally when the socket is disconnected already
BlockingConnection._handle_read()If the ReadPoller says there is data to read, try adn read it in the _handle_read of the parent class. Onceread, reset the counter that keeps track of how many frames have been written since the last read.
BlockingConnection._handle_timeout()Invoked whenever the socket times out
BlockingConnection._flush_outbound()Flush the outbound socket buffer.
BlockingConnection._on_connection_closed(method_frame, from_adapter=False)Called when the connection is closed remotely. The from_adapter value will be true if the connectionadapter has been disconnected from the broker and the method was invoked directly instead of by receivinga Connection.Close frame.
Parameters
• pika.frame.Method – The Connection.Close frame
• from_adapter (bool) – Called by the connection adapter
Raises AMQPConnectionError
BlockingConnection._send_frame(frame_value)This appends the fully generated frame to send to the broker to the output buffer which will be then sentvia the connection adapter.
Parameters frame_value (pika.frame.Frame|pika.frame.ProtocolHeader) – The frame to write
4.1.4 select_connection
A connection adapter that tries to use the best polling method for the platform pika is running on.
SelectConnection
class pika.adapters.select_connection.SelectConnection(parameters=None,on_open_callback=None,stop_ioloop_on_close=True)
An asynchronous connection adapter that attempts to use the fastest event loop adapter for the given platform.
SelectConnection._adapter_connect()Connect to the RabbitMQ broker
SelectConnection._flush_outbound()Call the state manager who will figure out that we need to write then call the poller’s poll function to forceit to process events.
IOLoop
class pika.adapters.select_connection.IOLoop(state_manager)Singlton wrapper that decides which type of poller to use, creates an instance of it in start_poller and keeps the
62 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
invoking application in a blocking state by calling the pollers start method. Poller should keep looping untilIOLoop.instance().stop() is called or there is a socket error.
Also provides a convenient pass-through for add_timeout and set_events
add_timeout(deadline, handler)Add a timeout with with given deadline, should return a timeout id.
Pass through a deadline and handler to the active poller.
Parameters
• deadline (int) – The number of seconds to wait until calling handler
• handler (method) – The method to call at deadline
Return type int
poller_typeReturn the type of poller.
Return type str
remove_timeout(timeout_id)Remove a timeout if it’s still in the timeout stack of the poller
Parameters timeout_id (str) – The timeout id to remove
start()Start the IOLoop, waiting for a Poller to take over.
start_poller(handler, events, fileno)Start the Poller, once started will take over for IOLoop.start()
Parameters
• handler (method) – The method to call to handle events
• events (int) – The events to handle
• fileno (int) – The file descriptor to poll for
stop()Stop the poller’s event loop
update_handler(fileno, events)Pass in the events to process for the given file descriptor.
Parameters
• fileno (int) – The file descriptor to poll for
• events (int) – The events to handle
SelectPoller
class pika.adapters.select_connection.SelectPoller(fileno, handler, events,state_manager)
Default behavior is to use Select since it’s the widest supported and has all of the methods we need for childclasses as well. One should only need to override the update_handler and start methods for additional types.
TIMEOUT = 1
add_timeout(deadline, handler)Add a timeout with with given deadline, should return a timeout id.
4.1. adapters 63
pika Documentation, Release 0.9.9
Parameters
• deadline (int) – The number of seconds to wait until calling handler
• handler (method) – The method to call at deadline
Return type str
flush_pending_timeouts()
poll(write_only=False)Check to see if the events that are cared about have fired.
Parameters write_only (bool) – Don’t look at self.events, just look to see if the adapter canwrite.
process_timeouts()Process the self._timeouts event stack
remove_timeout(timeout_id)Remove a timeout if it’s still in the timeout stack
Parameters timeout_id (str) – The timeout id to remove
start()Start the main poller loop. It will loop here until self.closed
update_handler(fileno, events)Set the events to the current events
Parameters
• fileno (int) – The file descriptor
• events (int) – The event mask
KQueuePoller
class pika.adapters.select_connection.KQueuePoller(fileno, handler, events,state_manager)
KQueuePoller works on BSD based systems and is faster than select
update_handler(fileno, events)Set the events to the current events
Parameters
• fileno (int) – The file descriptor
• events (int) – The event mask
start()Start the main poller loop. It will loop here until self.closed
poll(write_only=False)Check to see if the events that are cared about have fired.
Parameters write_only (bool) – Don’t look at self.events, just look to see if the adapter canwrite.
TIMEOUT = 1
add_timeout(deadline, handler)Add a timeout with with given deadline, should return a timeout id.
64 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Parameters
• deadline (int) – The number of seconds to wait until calling handler
• handler (method) – The method to call at deadline
Return type str
flush_pending_timeouts()
process_timeouts()Process the self._timeouts event stack
remove_timeout(timeout_id)Remove a timeout if it’s still in the timeout stack
Parameters timeout_id (str) – The timeout id to remove
PollPoller
class pika.adapters.select_connection.PollPoller(fileno, handler, events, state_manager)Poll works on Linux and can have better performance than EPoll in certain scenarios. Both are faster than select.
update_handler(fileno, events)Set the events to the current events
Parameters
• fileno (int) – The file descriptor
• events (int) – The event mask
start()Start the main poller loop. It will loop here until self.closed
poll(write_only=False)Poll until TIMEOUT waiting for an event
Parameters write_only (bool) – Only process write events
TIMEOUT = 1
add_timeout(deadline, handler)Add a timeout with with given deadline, should return a timeout id.
Parameters
• deadline (int) – The number of seconds to wait until calling handler
• handler (method) – The method to call at deadline
Return type str
flush_pending_timeouts()
process_timeouts()Process the self._timeouts event stack
remove_timeout(timeout_id)Remove a timeout if it’s still in the timeout stack
Parameters timeout_id (str) – The timeout id to remove
4.1. adapters 65
pika Documentation, Release 0.9.9
EPollPoller
class pika.adapters.select_connection.EPollPoller(fileno, handler, events, state_manager)EPoll works on Linux and can have better performance than Poll in certain scenarios. Both are faster than select.
poll(write_only=False)Poll until TIMEOUT waiting for an event
Parameters write_only (bool) – Only process write events
TIMEOUT = 1
add_timeout(deadline, handler)Add a timeout with with given deadline, should return a timeout id.
Parameters
• deadline (int) – The number of seconds to wait until calling handler
• handler (method) – The method to call at deadline
Return type str
flush_pending_timeouts()
process_timeouts()Process the self._timeouts event stack
remove_timeout(timeout_id)Remove a timeout if it’s still in the timeout stack
Parameters timeout_id (str) – The timeout id to remove
start()Start the main poller loop. It will loop here until self.closed
update_handler(fileno, events)Set the events to the current events
Parameters
• fileno (int) – The file descriptor
• events (int) – The event mask
4.1.5 tornado_connection
Run pika on the Tornado IOLoop
TornadoConnection
class pika.adapters.tornado_connection.TornadoConnection(parameters=None,on_open_callback=None,stop_ioloop_on_close=False,custom_ioloop=None)
The TornadoConnection runs on the Tornado IOLoop. If you’re running the connection in a web app, make sureyou set stop_ioloop_on_close to False, which is the default behavior for this adapter, otherwise the web app willstop taking requests.
TornadoConnection.WARN_ABOUT_IOLOOP = True
66 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
TornadoConnection._adapter_connect()Connect to the RabbitMQ broker
TornadoConnection._adapter_disconnect()Disconnect from the RabbitMQ broker
TornadoConnection.add_timeout(deadline, callback_method)Add the callback_method to the IOLoop timer to fire after deadline seconds. Returns a handle to thetimeout. Do not confuse with Tornado’s timeout where you pass in the time you want to have your callbackcalled. Only pass in the seconds until it’s to be called.
Parameters
• deadline (int) – The number of seconds to wait to call callback
• callback_method (method) – The callback method
Return type str
TornadoConnection.remove_timeout(timeout_id)Remove the timeout from the IOLoop by the ID returned from add_timeout.
Return type str
4.2 amqp_object
Note: This class is not intended for use by those using Pika in their applications. This documentation is for thosewho are extending Pika or otherwise working on the driver itself.
The class in this module, amqp_object.AMQPObject extends Python’s object class creating a base class that otherclasses, who would like an easy to implement representation of the class state, may extend. Base classes that areextended by low level AMQP frames and higher level AMQP classes and methods.
4.2.1 object
class pika.amqp_object.AMQPObjectBase object that is extended by AMQP low level frames and AMQP classes and methods.
NAME = ‘AMQPObject’
INDEX = None
4.2.2 Class
class pika.amqp_object.ClassIs extended by AMQP classes
NAME = ‘Unextended Class’
INDEX = None
4.2. amqp_object 67
pika Documentation, Release 0.9.9
4.2.3 Method
class pika.amqp_object.MethodIs extended by AMQP methods
NAME = ‘Unextended Method’
synchronous = False
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters
• properties (pika.frame.Properties) – AMQP Basic Properties
• body (str or unicode) – The message body
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
get_body()Return the message body if it is set.
Return type str|unicode
INDEX = None
4.2.4 Properties
class pika.amqp_object.PropertiesClass to encompass message properties (AMQP Basic.Properties)
NAME = ‘Unextended Properties’
INDEX = None
4.3 callback
Note: The following class is not intended for use by those using Pika in their applications. This documentation is forthose who are extending Pika or otherwise working on the driver itself.
Callback management class, common area for keeping track of all callbacks in the Pika stack.
4.3.1 CallbackManager
class pika.callback.CallbackManagerCallbackManager is a global callback system designed to be a single place where Pika can manage callbacksand process them. It should be referenced by the CallbackManager.instance() method instead of constructingnew instances of it.
add(*args, **kwargs)Add a callback to the stack for the specified key. If the call is specified as one_shot, it will be removedafter being fired
68 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
The prefix is usually the channel number but the class is generic and prefix and key may be any value.If you pass in only_caller CallbackManager will restrict processing of the callback to only the callingfunction/object that you specify.
Parameters
• prefix (str or int) – Categorize the callback
• key (object or str or dict) – The key for the callback
• callback (method) – The callback to call
• one_shot (bool) – Remove this callback after it is called
• only_caller (object) – Only allow one_caller value to call the event that fires the callback.
• arguments (dict) – Arguments to validate when processing
Return type tuple(prefix, key)
cleanup(*args, **kwargs)Remove all callbacks from the stack by a prefix. Returns True if keys were there to be removed
Parameters or int prefix (str) – The prefix for keeping track of callbacks with
Return type bool
clear()Clear all the callbacks if there are any defined.
pending(*args, **kwargs)Return count of callbacks for a given prefix or key or None
Parameters
• prefix (str or int) – Categorize the callback
• key (object or str or dict) – The key for the callback
Return type None or int
process(*args, **kwargs)Run through and process all the callbacks for the specified keys. Caller should be specified at all times sothat callbacks which require a specific function to call CallbackManager.process will not be processed.
Parameters
• prefix (str or int) – Categorize the callback
• key (object or str or dict) – The key for the callback
• caller (Object) – Who is firing the event
• args (list) – Any optional arguments
• keywords (dict) – Optional keyword arguments
Return type bool
remove(*args, **kwargs)Remove a callback from the stack by prefix, key and optionally the callback itself. If you only pass inprefix and key, all callbacks for that prefix and key will be removed.
Parameters
• prefix (str) – The prefix for keeping track of callbacks with
• key (str) – The callback key
4.3. callback 69
pika Documentation, Release 0.9.9
• callback_value (method) – The method defined to call on callback
• arguments (dict) – Optional arguments to check
Return type bool
remove_all(*args, **kwargs)Remove all callbacks for the specified prefix and key.
Parameters
• prefix (str) – The prefix for keeping track of callbacks with
• key (str) – The callback key
4.4 channel
The Channel class provides a wrapper for interacting with RabbitMQ implementing the methods and behaviors for anAMQP Channel.
4.4.1 Channel
class pika.channel.Channel(connection, channel_number, on_open_callback=None)A Channel is the primary communication method for interacting with RabbitMQ. It is recommended that youdo not directly invoke the creation of a channel object in your application code but rather construct the a channelby calling the active connection’s channel() method.
Channel.CLOSED = 0
Channel.OPENING = 1
Channel.OPEN = 2
Channel.CLOSING = 3
Channel.add_callback(callback, replies, one_shot=True)Pass in a callback handler and a list replies from the RabbitMQ broker which you’d like the callbacknotified of. Callbacks should allow for the frame parameter to be passed in.
Parameters
• callback (method) – The method to call
• replies (list) – The replies to get a callback for
• one_shot (bool) – Only handle the first type callback
Channel.add_on_cancel_callback(callback)Pass a callback function that will be called when the basic_cancel is sent by the server. The callbackfunction should receive a frame parameter.
Parameters callback (method) – The method to call on callback
Channel.add_on_close_callback(callback)Pass a callback function that will be called when the channel is closed. The callback function shouldreceive a frame parameter.
Parameters callback (method) – The method to call on callback
70 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Channel.add_on_flow_callback(callback)Pass a callback function that will be called when Channel.Flow is called by the remote server. Note thatnewer versions of RabbitMQ will not issue this but instead use TCP backpressure
Parameters callback (method) – The method to call on callback
Channel.add_on_return_callback(callback)Pass a callback function that will be called when basic_publish as sent a message that has been rejectedand returned by the server. The callback handler should receive a method, header and body frame. Thebase signature for the callback should be the same as the method signature one creates for a basic_consumecallback.
Parameters callback (method) – The method to call on callback
Channel.basic_ack(delivery_tag=0, multiple=False)Acknowledge one or more messages. When sent by the client, this method acknowledges one or more mes-sages delivered via the Deliver or Get-Ok methods. When sent by server, this method acknowledges one ormore messages published with the Publish method on a channel in confirm mode. The acknowledgementcan be for a single message or a set of messages up to and including a specific message.
Parameters
• delivery-tag (int) – The server-assigned delivery tag
• multiple (bool) – If set to True, the delivery tag is treated as “up to and including”, so thatmultiple messages can be acknowledged with a single method. If set to False, the deliverytag refers to a single message. If the multiple field is 1, and the delivery tag is zero, thisindicates acknowledgement of all outstanding messages.
Channel.basic_cancel(callback=None, consumer_tag=’‘, nowait=False)This method cancels a consumer. This does not affect already delivered messages, but it does mean theserver will not send any more messages for that consumer. The client may receive an arbitrary numberof messages in between sending the cancel method and receiving the cancel-ok reply. It may also be sentfrom the server to the client in the event of the consumer being unexpectedly cancelled (i.e. cancelled forany reason other than the server receiving the corresponding basic.cancel from the client). This allowsclients to be notified of the loss of consumers due to events such as queue deletion.
Parameters
• callback (method) – Method to call for a Basic.CancelOk response
• consumer_tag (str) – Identifier for the consumer
• nowait (bool) – Do not expect a Basic.CancelOk response
Raises ValueError
Channel.basic_consume(consumer_callback, queue=’‘, no_ack=False, exclusive=False, con-sumer_tag=None)
Sends the AMQP command Basic.Consume to the broker and binds messages for the consumer_tag to theconsumer callback. If you do not pass in a consumer_tag, one will be automatically generated for you.Returns the consumer tag.
For more information on basic_consume, see: http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.consume
Parameters
• consumer_callback (method) – The method to callback when consuming
• queue (str or unicode) – The queue to consume from
• no_ack (bool) – Tell the broker to not expect a response
4.4. channel 71
pika Documentation, Release 0.9.9
• exclusive (bool) – Don’t allow other consumers on the queue
• consumer_tag (str or unicode) – Specify your own consumer tag
Return type str
Channel.basic_get(callback=None, queue=’‘, no_ack=False)Get a single message from the AMQP broker. The callback method signature should have 3 parameters:The method frame, header frame and the body, like the consumer callback for Basic.Consume. If you wantto be notified of Basic.GetEmpty, use the Channel.add_callback method adding your Basic.GetEmptycallback which should expect only one parameter, frame. For more information on basic_get and itsparameters, see:
http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.get
Parameters
• callback (method) – The method to callback with a message
• queue (str or unicode) – The queue to get a message from
• no_ack (bool) – Tell the broker to not expect a reply
Channel.basic_nack(delivery_tag=None, multiple=False, requeue=True)This method allows a client to reject one or more incoming messages. It can be used to interrupt and cancellarge incoming messages, or return untreatable messages to their original queue.
Parameters
• delivery-tag (int) – The server-assigned delivery tag
• multiple (bool) – If set to True, the delivery tag is treated as “up to and including”, so thatmultiple messages can be acknowledged with a single method. If set to False, the deliverytag refers to a single message. If the multiple field is 1, and the delivery tag is zero, thisindicates acknowledgement of all outstanding messages.
• requeue (bool) – If requeue is true, the server will attempt to requeue the message. Ifrequeue is false or the requeue attempt fails the messages are discarded or dead-lettered.
Channel.basic_publish(exchange, routing_key, body, properties=None, mandatory=False, im-mediate=False)
Publish to the channel with the given exchange, routing key and body. For more information on ba-sic_publish and what the parameters do, see:
http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.publish
Parameters
• exchange (str or unicode) – The exchange to publish to
• routing_key (str or unicode) – The routing key to bind on
• body (str or unicode) – The message body
• properties (pika.spec.Properties) – Basic.properties
• mandatory (bool) – The mandatory flag
• immediate (bool) – The immediate flag
Channel.basic_qos(callback=None, prefetch_size=0, prefetch_count=0, all_channels=False)Specify quality of service. This method requests a specific quality of service. The QoS can be specifiedfor the current channel or for all channels on the connection. The client can request that messages be sentin advance so that when the client finishes processing a message, the following message is already heldlocally, rather than needing to be sent down the channel. Prefetching gives a performance improvement.
72 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Parameters
• callback (method) – The method to callback for Basic.QosOk response
• prefetch_size (int) – This field specifies the prefetch window size. The server will senda message in advance if it is equal to or smaller in size than the available prefetch size(and also falls into other prefetch limits). May be set to zero, meaning “no specific limit”,although other prefetch limits may still apply. The prefetch-size is ignored if the no-ackoption is set.
• prefetch_count (int) – Specifies a prefetch window in terms of whole messages. Thisfield may be used in combination with the prefetch-size field; a message will only be sentin advance if both prefetch windows (and those at the channel and connection level) allowit. The prefetch-count is ignored if the no-ack option is set.
• all_channels (bool) – Should the QoS apply to all channels
Channel.basic_reject(delivery_tag=None, requeue=True)Reject an incoming message. This method allows a client to reject a message. It can be used to interruptand cancel large incoming messages, or return untreatable messages to their original queue.
Parameters
• delivery-tag (int) – The server-assigned delivery tag
• requeue (bool) – If requeue is true, the server will attempt to requeue the message. Ifrequeue is false or the requeue attempt fails the messages are discarded or dead-lettered.
Channel.basic_recover(callback=None, requeue=False)This method asks the server to redeliver all unacknowledged messages on a specified channel. Zero ormore messages may be redelivered. This method replaces the asynchronous Recover.
Parameters
• callback (method) – Method to call when receiving Basic.RecoverOk
• requeue (bool) – If False, the message will be redelivered to the original recipient. IfTrue, the server will attempt to requeue the message, potentially then delivering it to analternative subscriber.
Channel.close(reply_code=0, reply_text=’Normal Shutdown’)Will invoke a clean shutdown of the channel with the AMQP Broker.
Parameters
• reply_code (int) – The reply code to close the channel with
• reply_text (str) – The reply text to close the channel with
Channel.confirm_delivery(callback=None, nowait=False)Turn on Confirm mode in the channel. Pass in a callback to be notified by the Broker when a message hasbeen confirmed as received or rejected (Basic.Ack, Basic.Nack) from the broker to the publisher.
For more information see: http://www.rabbitmq.com/extensions.html#confirms
Parameters
• callback (method) – The callback for delivery confirmations
• nowait (bool) – Do not send a reply frame (Confirm.SelectOk)
Channel.consumer_tagsProperty method that returns a list of currently active consumers
Return type list
4.4. channel 73
pika Documentation, Release 0.9.9
Channel.exchange_bind(callback=None, destination=None, source=None, routing_key=’‘,nowait=False, arguments=None)
Bind an exchange to another exchange.
Parameters
• callback (method) – The method to call on Exchange.BindOk
• destination (str or unicode) – The destination exchange to bind
• source (str or unicode) – The source exchange to bind to
• routing_key (str or unicode) – The routing key to bind on
• nowait (bool) – Do not wait for an Exchange.BindOk
• arguments (dict) – Custom key/value pair arguments for the binding
Channel.exchange_declare(callback=None, exchange=None, exchange_type=’direct’, pas-sive=False, durable=False, auto_delete=False, internal=False,nowait=False, arguments=None, type=None)
This method creates an exchange if it does not already exist, and if the exchange exists, verifies that it isof the correct and expected class.
If passive set, the server will reply with Declare-Ok if the exchange already exists with the same name, andraise an error if not and if the exchange does not already exist, the server MUST raise a channel exceptionwith reply code 404 (not found).
Parameters
• callback (method) – Call this method on Exchange.DeclareOk
• exchange (str or unicode sequence of these characters: letters, digits, hyphen, underscore,period, or colon.) – The exchange name consists of a non-empty
• exchange_type (str) – The exchange type to use
• passive (bool) – Perform a declare or just check to see if it exists
• durable (bool) – Survive a reboot of RabbitMQ
• auto_delete (bool) – Remove when no more queues are bound to it
• internal (bool) – Can only be published to by other exchanges
• nowait (bool) – Do not expect an Exchange.DeclareOk response
• arguments (dict) – Custom key/value pair arguments for the exchange
• type (str) – The deprecated exchange type parameter
Channel.exchange_delete(callback=None, exchange=None, if_unused=False, nowait=False)Delete the exchange.
Parameters
• callback (method) – The method to call on Exchange.DeleteOk
• exchange (str or unicode) – The exchange name
• if_unused (bool) – only delete if the exchange is unused
• nowait (bool) – Do not wait for an Exchange.DeleteOk
Channel.exchange_unbind(callback=None, destination=None, source=None, routing_key=’‘,nowait=False, arguments=None)
Unbind an exchange from another exchange.
74 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Parameters
• callback (method) – The method to call on Exchange.UnbindOk
• destination (str or unicode) – The destination exchange to unbind
• source (str or unicode) – The source exchange to unbind from
• routing_key (str or unicode) – The routing key to unbind
• nowait (bool) – Do not wait for an Exchange.UnbindOk
• arguments (dict) – Custom key/value pair arguments for the binding
Channel.flow(callback, active)Turn Channel flow control off and on. Pass a callback to be notified of the response from the server. activeis a bool. Callback should expect a bool in response indicating channel flow state. For more information,please reference:
http://www.rabbitmq.com/amqp-0-9-1-reference.html#channel.flow
Parameters
• callback (method) – The callback method
• active (bool) – Turn flow on or off
Channel.is_closedReturns True if the channel is closed.
Return type bool
Channel.is_closingReturns True if the channel is closing.
Return type bool
Channel.is_openReturns True if the channel is open.
Return type bool
Channel.open()Open the channel
Channel.queue_bind(callback, queue, exchange, routing_key=None, nowait=False, argu-ments=None)
Bind the queue to the specified exchange
Parameters
• callback (method) – The method to call on Queue.BindOk
• queue (str or unicode) – The queue to bind to the exchange
• exchange (str or unicode) – The source exchange to bind to
• routing_key (str or unicode) – The routing key to bind on
• nowait (bool) – Do not wait for a Queue.BindOk
• arguments (dict) – Custom key/value pair arguments for the binding
Channel.queue_declare(callback, queue=’‘, passive=False, durable=False, exclusive=False,auto_delete=False, nowait=False, arguments=None)
Declare queue, create if needed. This method creates or checks a queue. When creating a new queue theclient can specify various properties that control the durability of the queue and its contents, and the levelof sharing for the queue.
4.4. channel 75
pika Documentation, Release 0.9.9
Leave the queue name empty for a auto-named queue in RabbitMQ
Parameters
• callback (method) – The method to call on Queue.DeclareOk
• queue (str or unicode) – The queue name
• passive (bool) – Only check to see if the queue exists
• durable (bool) – Survive reboots of the broker
• exclusive (bool) – Only allow access by the current connection
• auto_delete (bool) – Delete after consumer cancels or disconnects
• nowait (bool) – Do not wait for a Queue.DeclareOk
• arguments (dict) – Custom key/value arguments for the queue
Channel.queue_delete(callback=None, queue=’‘, if_unused=False, if_empty=False,nowait=False)
Delete a queue from the broker.
Parameters
• callback (method) – The method to call on Queue.DeleteOk
• queue (str or unicode) – The queue to delete
• if_unused (bool) – only delete if it’s unused
• if_empty (bool) – only delete if the queue is empty
• nowait (bool) – Do not wait for a Queue.DeleteOk
Channel.queue_purge(callback=None, queue=’‘, nowait=False)Purge all of the messages from the specified queue
Parameters
• callback (method) – The method to call on Queue.PurgeOk
• queue (str or unicode) – The queue to purge
• nowait (bool) – Do not expect a Queue.PurgeOk response
Channel.queue_unbind(callback=None, queue=’‘, exchange=None, routing_key=’‘, argu-ments=None)
Unbind a queue from an exchange.
Parameters
• callback (method) – The method to call on Queue.UnbindOk
• queue (str or unicode) – The queue to unbind from the exchange
• exchange (str or unicode) – The source exchange to bind from
• routing_key (str or unicode) – The routing key to unbind
• arguments (dict) – Custom key/value pair arguments for the binding
Channel.tx_commit(callback=None)Commit a transaction
Parameters callback (method) – The callback for delivery confirmations
Channel.tx_rollback(callback=None)Rollback a transaction.
76 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Parameters callback (method) – The callback for delivery confirmations
Channel.tx_select(callback=None)Select standard transaction mode. This method sets the channel to use standard transactions. The clientmust use this method at least once on a channel before using the Commit or Rollback methods.
Parameters callback (method) – The callback for delivery confirmations
Channel._add_callbacks()Callbacks that add the required behavior for a channel when connecting and connected to a server.
Channel._add_pending_msg(consumer_tag, method_frame, header_frame, body)Add the received message to the pending message stack.
Parameters
• consumer_tag (str) – The consumer tag for the message
• method_frame (pika.frame.Method) – The received method frame
• header_frame (pika.frame.Header) – The received header frame
• body (str or unicode) – The message body
Channel._cleanup()Remove any callbacks for the channel.
Channel._get_pending_msg(consumer_tag)Get a pending message for the consumer tag from the stack.
Parameters consumer_tag (str) – The consumer tag to get a message from
Return type tuple(pika.frame.Header, pika.frame.Method, str|unicode)
Channel._handle_content_frame(frame_value)This is invoked by the connection when frames that are not registered with the CallbackManager have beenfound. This should only be the case when the frames are related to content delivery.
The frame_dispatcher will be invoked which will return the fully formed message in three parts when allof the body frames have been received.
Parameters frame_value (pika.amqp_object.Frame) – The frame to deliver
Channel._has_content(method_frame)Return a bool if it’s a content method as defined by the spec
Parameters method_frame (pika.amqp_object.Method) – The method frame received
Channel._on_cancel(method_frame)When the broker cancels a consumer, delete it from our internal dictionary.
Parameters method_frame (pika.frame.Method) – The method frame received
Channel._on_cancelok(method_frame)
Called in response to a frame from the Broker when the client sends Basic.Cancel
Parameters method_frame (pika.frame.Method) – The method frame received
Channel._on_close(method_frame)Handle the case where our channel has been closed for us
Parameters method_frame (pika.frame.Method) – The close frame
4.4. channel 77
pika Documentation, Release 0.9.9
Channel._on_deliver(method_frame, header_frame, body)Cope with reentrancy. If a particular consumer is still active when another delivery appears for it, queuethe deliveries up until it finally exits.
Parameters
• method_frame (pika.frame.Method) – The method frame received
• header_frame (pika.frame.Header) – The header frame received
• body (str or unicode) – The body received
Channel._on_eventok(method_frame)Generic events that returned ok that may have internal callbacks. We keep a list of what we’ve yet toimplement so that we don’t silently drain events that we don’t support.
Parameters method_frame (pika.frame.Method) – The method frame received
Channel._on_flow(method_frame_unused)Called if the server sends a Channel.Flow frame.
Parameters method_frame_unused (pika.frame.Method) – The Channel.Flow frame
Channel._on_flowok(method_frame)Called in response to us asking the server to toggle on Channel.Flow
Parameters method_frame (pika.frame.Method) – The method frame received
Channel._on_getempty(method_frame)When we receive an empty reply do nothing but log it
Parameters method_frame (pika.frame.Method) – The method frame received
Channel._on_getok(method_frame, header_frame, body)Called in reply to a Basic.Get when there is a message.
Parameters
• method_frame (pika.frame.Method) – The method frame received
• header_frame (pika.frame.Header) – The header frame received
• body (str or unicode) – The body received
Channel._on_openok(frame_unused)Called by our callback handler when we receive a Channel.OpenOk and subsequently calls our_on_openok_callback which was passed into the Channel constructor. The reason we do this is becausewe want to make sure that the on_open_callback parameter passed into the Channel constructor is not thefirst callback we make.
Parameters frame_unused (pika.frame.Method) – Unused Channel.OpenOk frame
Channel._on_return(method_frame, header_frame, body)Called if the server sends a Basic.Return frame.
Parameters
• method_frame (pika.frame.Method) – The Basic.Return frame
• header_frame (pika.frame.Header) – The content header frame
• body (str or unicode) – The message body
Channel._on_selectok(method_frame)Called when the broker sends a Confirm.SelectOk frame
Parameters method_frame (pika.frame.Method) – The method frame received
78 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Channel._on_synchronous_complete(method_frame_unused)This is called when a synchronous command is completed. It will undo the blocking state and send all theframes that stacked up while we were in the blocking state.
Parameters method_frame_unused (pika.frame.Method) – The method frame received
Channel._rpc(method_frame, callback=None, acceptable_replies=None)Shortcut wrapper to the Connection’s rpc command using its callback stack, passing in our channel number.
Parameters
• method_frame (pika.amqp_object.Method) – The method frame to call
• callback (method) – The callback for the RPC response
• acceptable_replies (list) – The replies this RPC call expects
Channel._send_method(method_frame, content=None)Shortcut wrapper to send a method through our connection, passing in the channel number
Parameters
• method_frame (pika.object.Method) – The method frame to send
• content (tuple) – If set, is a content frame, is tuple of properties and body.
Channel._set_state(connection_state)Set the channel connection state to the specified state value.
Parameters connection_state (int) – The connection_state value
Channel._shutdown()Called when close() is invoked either directly or when all of the consumers have been cancelled.
Channel._unexpected_frame(frame_value)
Channel._validate_channel_and_callback(callback)
4.4.2 ContentFrameDispatcher
class pika.channel.ContentFrameDispatcherHandle content related frames, building a message and return the message back in three parts upon receipt.
process(frame_value)Invoked by the Channel object when passed frames that are not setup in the rpc process and that don’t haveexplicit reply types defined. This includes Basic.Publish, Basic.GetOk and Basic.Return
Parameters frame_value (Method|Header|Body) – The frame to process
_finish()Invoked when all of the message has been received
Return type tuple(pika.frame.Method, pika.frame.Header, str|unicode)
_handle_body_frame(body_frame)Receive body frames and append them to the stack. When the body size matches, call the finish method.
Parameters body_frame (Body) – The body frame
Raises pika.exceptions.BodyTooLongError
Return type tuple(pika.frame.Method, pika.frame.Header, str|unicode)|None
_reset()Reset the values for processing frames
4.4. channel 79
pika Documentation, Release 0.9.9
4.5 connection
Note: The following class level documentation is not intended for use by those using Pika in their applications. Thisdocumentation is for those who are extending Pika or otherwise working on the driver itself. For an overview of howto use adapters, please reference the Connecting to RabbitMQ documentation.
Core connection objects
4.5.1 Parameters
class pika.connection.ParametersBase connection parameters class definition
Parameters
• DEFAULT_HOST (str) – ‘localhost’
• DEFAULT_PORT (int) – 5672
• DEFAULT_VIRTUAL_HOST (str) – ‘/’
• DEFAULT_USERNAME (str) – ‘guest’
• DEFAULT_PASSWORD (str) – ‘guest’
• DEFAULT_HEARTBEAT_INTERVAL (int) – 0
• DEFAULT_CHANNEL_MAX (int) – 0
• DEFAULT_FRAME_MAX (int) – pika.spec.FRAME_MAX_SIZE
• DEFAULT_LOCALE (str) – ‘en_US’
• DEFAULT_CONNECTION_ATTEMPTS (int) – 1
• DEFAULT_RETRY_DELAY (int|float) – 2.0
• DEFAULT_SOCKET_TIMEOUT (int|float) – 0.25
• DEFAULT_SSL (bool) – False
• DEFAULT_SSL_OPTIONS (dict) – {}
• DEFAULT_SSL_PORT (int) – 5671
• DEFAULT_BACKPRESSURE_DETECTION (bool) – False
Parameters.DEFAULT_BACKPRESSURE_DETECTION = False
Parameters.DEFAULT_CONNECTION_ATTEMPTS = 1
Parameters.DEFAULT_CHANNEL_MAX = 0
Parameters.DEFAULT_FRAME_MAX = 131072
Parameters.DEFAULT_HEARTBEAT_INTERVAL = 0
Parameters.DEFAULT_HOST = ‘localhost’
Parameters.DEFAULT_LOCALE = ‘en_US’
Parameters.DEFAULT_PASSWORD = ‘guest’
Parameters.DEFAULT_PORT = 5672
80 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Parameters.DEFAULT_RETRY_DELAY = 2.0
Parameters.DEFAULT_SOCKET_TIMEOUT = 0.25
Parameters.DEFAULT_SSL = False
Parameters.DEFAULT_SSL_OPTIONS = {}
Parameters.DEFAULT_SSL_PORT = 5671
Parameters.DEFAULT_USERNAME = ‘guest’
Parameters.DEFAULT_VIRTUAL_HOST = ‘/’
Parameters._credentials(username, password)Return a plain credentials object for the specified username and password.
Parameters
• username (str) – The username to use
• password (str) – The password to use
Return type pika_credentials.PlainCredentials
Parameters._validate_backpressure(backpressure_detection)Validate that the backpressure detection option is a bool.
Parameters backpressure_detection (bool) – The backpressure detection value
Return type bool
Raises TypeError
Parameters._validate_channel_max(channel_max)Validate that the channel_max value is an int
Parameters channel_max (int) – The value to validate
Return type bool
Raises TypeError
Raises ValueError
Parameters._validate_connection_attempts(connection_attempts)Validate that the channel_max value is an int
Parameters connection_attempts (int) – The value to validate
Return type bool
Raises TypeError
Raises ValueError
Parameters._validate_credentials(credentials)Validate the credentials passed in are using a valid object type.
Parameters credentials (pika.credentials.Credentials) – Credentials to validate
Return type bool
Raises TypeError
Parameters._validate_frame_max(frame_max)
Validate that the frame_max value is an int and does not exceed the maximum frame size and is notless than the frame min size.
4.5. connection 81
pika Documentation, Release 0.9.9
Parameters frame_max (int) – The value to validate
Return type bool
Raises TypeError
Raises InvalidMinimumFrameSize
Parameters._validate_heartbeat_interval(heartbeat_interval)Validate that the heartbeat_interval value is an int
Parameters heartbeat_interval (int) – The value to validate
Return type bool
Raises TypeError
Raises ValueError
Parameters._validate_host(host)Validate that the host value is an str
Parameters host (str|unicode) – The value to validate
Return type bool
Raises TypeError
Parameters._validate_locale(locale)Validate that the locale value is an str
Parameters locale (str) – The value to validate
Return type bool
Raises TypeError
Parameters._validate_port(port)Validate that the port value is an int
Parameters port (int) – The value to validate
Return type bool
Raises TypeError
Parameters._validate_retry_delay(retry_delay)Validate that the retry_delay value is an int or float
Parameters retry_delay (int|float) – The value to validate
Return type bool
Raises TypeError
Parameters._validate_socket_timeout(socket_timeout)Validate that the socket_timeout value is an int or float
Parameters socket_timeout (int|float) – The value to validate
Return type bool
Raises TypeError
Parameters._validate_ssl(ssl)Validate the SSL toggle is a bool
Parameters ssl (bool) – The SSL enabled/disabled value
82 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Return type bool
Raises TypeError
Parameters._validate_ssl_options(ssl_options)Validate the SSL options value is a dictionary.
Parameters ssl_options (dict|None) – SSL Options to validate
Return type bool
Raises TypeError
Parameters._validate_virtual_host(virtual_host)Validate that the virtual_host value is an str
Parameters virtual_host (str) – The value to validate
Return type bool
Raises TypeError
4.5.2 ConnectionParameters
class pika.connection.ConnectionParameters(host=None, port=None, virtual_host=None,credentials=None, channel_max=None,frame_max=None, heartbeat_interval=None,ssl=None, ssl_options=None, connec-tion_attempts=None, retry_delay=None,socket_timeout=None, locale=None, back-pressure_detection=None)
Connection parameters object that is passed into the connection adapter upon construction.
ConnectionParameters.DEFAULT_BACKPRESSURE_DETECTION = False
ConnectionParameters.DEFAULT_CHANNEL_MAX = 0
ConnectionParameters.DEFAULT_CONNECTION_ATTEMPTS = 1
ConnectionParameters.DEFAULT_FRAME_MAX = 131072
ConnectionParameters.DEFAULT_HEARTBEAT_INTERVAL = 0
ConnectionParameters.DEFAULT_HOST = ‘localhost’
ConnectionParameters.DEFAULT_LOCALE = ‘en_US’
ConnectionParameters.DEFAULT_PASSWORD = ‘guest’
ConnectionParameters.DEFAULT_PORT = 5672
ConnectionParameters.DEFAULT_RETRY_DELAY = 2.0
ConnectionParameters.DEFAULT_SOCKET_TIMEOUT = 0.25
ConnectionParameters.DEFAULT_SSL = False
ConnectionParameters.DEFAULT_SSL_OPTIONS = {}
ConnectionParameters.DEFAULT_SSL_PORT = 5671
ConnectionParameters.DEFAULT_USERNAME = ‘guest’
ConnectionParameters.DEFAULT_VIRTUAL_HOST = ‘/’
4.5. connection 83
pika Documentation, Release 0.9.9
ConnectionParameters._credentials(username, password)Return a plain credentials object for the specified username and password.
Parameters
• username (str) – The username to use
• password (str) – The password to use
Return type pika_credentials.PlainCredentials
ConnectionParameters._validate_backpressure(backpressure_detection)Validate that the backpressure detection option is a bool.
Parameters backpressure_detection (bool) – The backpressure detection value
Return type bool
Raises TypeError
ConnectionParameters._validate_channel_max(channel_max)Validate that the channel_max value is an int
Parameters channel_max (int) – The value to validate
Return type bool
Raises TypeError
Raises ValueError
ConnectionParameters._validate_connection_attempts(connection_attempts)Validate that the channel_max value is an int
Parameters connection_attempts (int) – The value to validate
Return type bool
Raises TypeError
Raises ValueError
ConnectionParameters._validate_credentials(credentials)Validate the credentials passed in are using a valid object type.
Parameters credentials (pika.credentials.Credentials) – Credentials to validate
Return type bool
Raises TypeError
ConnectionParameters._validate_frame_max(frame_max)
Validate that the frame_max value is an int and does not exceed the maximum frame size and is notless than the frame min size.
Parameters frame_max (int) – The value to validate
Return type bool
Raises TypeError
Raises InvalidMinimumFrameSize
ConnectionParameters._validate_heartbeat_interval(heartbeat_interval)Validate that the heartbeat_interval value is an int
Parameters heartbeat_interval (int) – The value to validate
84 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Return type bool
Raises TypeError
Raises ValueError
ConnectionParameters._validate_host(host)Validate that the host value is an str
Parameters host (str|unicode) – The value to validate
Return type bool
Raises TypeError
ConnectionParameters._validate_locale(locale)Validate that the locale value is an str
Parameters locale (str) – The value to validate
Return type bool
Raises TypeError
ConnectionParameters._validate_port(port)Validate that the port value is an int
Parameters port (int) – The value to validate
Return type bool
Raises TypeError
ConnectionParameters._validate_retry_delay(retry_delay)Validate that the retry_delay value is an int or float
Parameters retry_delay (int|float) – The value to validate
Return type bool
Raises TypeError
ConnectionParameters._validate_socket_timeout(socket_timeout)Validate that the socket_timeout value is an int or float
Parameters socket_timeout (int|float) – The value to validate
Return type bool
Raises TypeError
ConnectionParameters._validate_ssl(ssl)Validate the SSL toggle is a bool
Parameters ssl (bool) – The SSL enabled/disabled value
Return type bool
Raises TypeError
ConnectionParameters._validate_ssl_options(ssl_options)Validate the SSL options value is a dictionary.
Parameters ssl_options (dict|None) – SSL Options to validate
Return type bool
Raises TypeError
4.5. connection 85
pika Documentation, Release 0.9.9
ConnectionParameters._validate_virtual_host(virtual_host)Validate that the virtual_host value is an str
Parameters virtual_host (str) – The value to validate
Return type bool
Raises TypeError
4.5.3 URLParameters
class pika.connection.URLParameters(url)Create a Connection parameters object based off of URIParameters
URLParameters._process_url(url)Take an AMQP URL and break it up into the various parameters.
Parameters url (str) – The URL to parse
URLParameters.DEFAULT_BACKPRESSURE_DETECTION = False
URLParameters.DEFAULT_CHANNEL_MAX = 0
URLParameters.DEFAULT_CONNECTION_ATTEMPTS = 1
URLParameters.DEFAULT_FRAME_MAX = 131072
URLParameters.DEFAULT_HEARTBEAT_INTERVAL = 0
URLParameters.DEFAULT_HOST = ‘localhost’
URLParameters.DEFAULT_LOCALE = ‘en_US’
URLParameters.DEFAULT_PASSWORD = ‘guest’
URLParameters.DEFAULT_PORT = 5672
URLParameters.DEFAULT_RETRY_DELAY = 2.0
URLParameters.DEFAULT_SOCKET_TIMEOUT = 0.25
URLParameters.DEFAULT_SSL = False
URLParameters.DEFAULT_SSL_OPTIONS = {}
URLParameters.DEFAULT_SSL_PORT = 5671
URLParameters.DEFAULT_USERNAME = ‘guest’
URLParameters.DEFAULT_VIRTUAL_HOST = ‘/’
URLParameters._credentials(username, password)Return a plain credentials object for the specified username and password.
Parameters
• username (str) – The username to use
• password (str) – The password to use
Return type pika_credentials.PlainCredentials
URLParameters._validate_backpressure(backpressure_detection)Validate that the backpressure detection option is a bool.
Parameters backpressure_detection (bool) – The backpressure detection value
Return type bool
86 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Raises TypeError
URLParameters._validate_channel_max(channel_max)Validate that the channel_max value is an int
Parameters channel_max (int) – The value to validate
Return type bool
Raises TypeError
Raises ValueError
URLParameters._validate_connection_attempts(connection_attempts)Validate that the channel_max value is an int
Parameters connection_attempts (int) – The value to validate
Return type bool
Raises TypeError
Raises ValueError
URLParameters._validate_credentials(credentials)Validate the credentials passed in are using a valid object type.
Parameters credentials (pika.credentials.Credentials) – Credentials to validate
Return type bool
Raises TypeError
URLParameters._validate_frame_max(frame_max)
Validate that the frame_max value is an int and does not exceed the maximum frame size and is notless than the frame min size.
Parameters frame_max (int) – The value to validate
Return type bool
Raises TypeError
Raises InvalidMinimumFrameSize
URLParameters._validate_heartbeat_interval(heartbeat_interval)Validate that the heartbeat_interval value is an int
Parameters heartbeat_interval (int) – The value to validate
Return type bool
Raises TypeError
Raises ValueError
URLParameters._validate_host(host)Validate that the host value is an str
Parameters host (str|unicode) – The value to validate
Return type bool
Raises TypeError
URLParameters._validate_locale(locale)Validate that the locale value is an str
4.5. connection 87
pika Documentation, Release 0.9.9
Parameters locale (str) – The value to validate
Return type bool
Raises TypeError
URLParameters._validate_port(port)Validate that the port value is an int
Parameters port (int) – The value to validate
Return type bool
Raises TypeError
URLParameters._validate_retry_delay(retry_delay)Validate that the retry_delay value is an int or float
Parameters retry_delay (int|float) – The value to validate
Return type bool
Raises TypeError
URLParameters._validate_socket_timeout(socket_timeout)Validate that the socket_timeout value is an int or float
Parameters socket_timeout (int|float) – The value to validate
Return type bool
Raises TypeError
URLParameters._validate_ssl(ssl)Validate the SSL toggle is a bool
Parameters ssl (bool) – The SSL enabled/disabled value
Return type bool
Raises TypeError
URLParameters._validate_ssl_options(ssl_options)Validate the SSL options value is a dictionary.
Parameters ssl_options (dict|None) – SSL Options to validate
Return type bool
Raises TypeError
URLParameters._validate_virtual_host(virtual_host)Validate that the virtual_host value is an str
Parameters virtual_host (str) – The value to validate
Return type bool
Raises TypeError
4.5.4 Connection
class pika.connection.Connection(parameters=None, on_open_callback=None)This is the core class that implements communication with RabbitMQ. This class should not be invoked directlybut rather through the use of an adapter such as SelectConnection or BlockingConnection.
Connection.CONNECTION_CLOSED = 0
88 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Connection.CONNECTION_INIT = 1
Connection.CONNECTION_PROTOCOL = 2
Connection.CONNECTION_START = 3
Connection.CONNECTION_TUNE = 4
Connection.CONNECTION_OPEN = 5
Connection.CONNECTION_CLOSING = 6
Connection.add_backpressure_callback(callback_method)Call method “callback” when pika believes backpressure is being applied.
Parameters callback_method (method) – The method to call
Connection.add_on_close_callback(callback_method)Add a callback notification when the connection has closed.
Parameters callback_method (method) – The callback when the channel is opened
Connection.add_on_open_callback(callback_method)Add a callback notification when the connection has opened.
Parameters callback_method (method) – The callback when the channel is opened
Connection.add_timeout(deadline, callback_method)Adapters should override to call the callback after the specified number of seconds have elapsed, using atimer, or a thread, or similar.
Parameters
• deadline (int) – The number of seconds to wait to call callback
• callback_method (method) – The callback method
Connection.channel(on_open_callback, channel_number=None)Create a new channel with the next available channel number or pass in a channel number to use. Must benon-zero if you would like to specify but it is recommended that you let Pika manage the channel numbers.
Parameters
• on_open_callback (method) – The callback when the channel is opened
• channel_number (int) – The channel number to use, defaults to the next available.
Return type pika.channel.Channel
Connection.close(reply_code=200, reply_text=’Normal shutdown’)Disconnect from RabbitMQ. If there are any open channels, it will attempt to close them prior to fullydisconnecting. Channels which have active consumers will attempt to send a Basic.Cancel to RabbitMQto cleanly stop the delivery of messages prior to closing the channel.
Parameters
• reply_code (int) – The code number for the close
• reply_text (str) – The text reason for the close
Connection.remove_timeout(callback_method)Adapters should override to call the callback after the specified number of seconds have elapsed, using atimer, or a thread, or similar.
Parameters callback_method (method) – The callback to remove a timeout for
4.5. connection 89
pika Documentation, Release 0.9.9
Connection.set_backpressure_multiplier(value=10)Alter the backpressure multiplier value. We set this to 10 by default. This value is used to raise warningsand trigger the backpressure callback.
Parameters value (int) – The multiplier value to set
Connection.is_closedReturns a boolean reporting the current connection state.
Connection.is_closingReturns a boolean reporting the current connection state.
Connection.is_openReturns a boolean reporting the current connection state.
Connection.basic_nackSpecifies if the server supports basic.nack on the active connection.
Return type bool
Connection.consumer_cancel_notifySpecifies if the server supports consumer cancel notification on the active connection.
Return type bool
Connection.exchange_exchange_bindingsSpecifies if the active connection supports exchange to exchange bindings.
Return type bool
Connection.publisher_confirmsSpecifies if the active connection can use publisher confirmations.
Return type bool
Connection._adapter_connect()Subclasses should override to set up the outbound socket connection.
Raises NotImplementedError
Connection._adapter_disconnect()Subclasses should override this to cause the underlying transport (socket) to close.
Raises NotImplementedError
Connection._add_channel_callbacks(channel_number)Add the appropriate callbacks for the specified channel number.
Parameters channel_number (int) – The channel number for the callbacks
Connection._add_connection_start_callback()Add a callback for when a Connection.Start frame is received from the broker.
Connection._add_connection_tune_callback()Add a callback for when a Connection.Tune frame is received.
Connection._append_frame_buffer(value)Append the bytes to the frame buffer.
Parameters value (str) – The bytes to append to the frame buffer
Connection._buffer_sizeReturn the suggested buffer size from the connection state/tune or the default if that is None.
Return type int
90 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Connection._check_for_protocol_mismatch(value)Invoked when starting a connection to make sure it’s a supported protocol.
Parameters value (pika.frame.Method) – The frame to check
Raises ProtocolVersionMismatch
Connection._client_propertiesReturn the client properties dictionary.
Return type dict
Connection._close_channels(reply_code, reply_text)Close the open channels with the specified reply_code and reply_text.
Parameters
• reply_code (int) – The code for why the channels are being closed
• reply_text (str) – The text reason for why the channels are closing
Connection._combine(a, b)Pass in two values, if a is 0, return b otherwise if b is 0, return a. If neither case matches return the smallestvalue.
Parameters
• a (int) – The first value
• b (int) – The second value
Return type int
Connection._connect()Call the Adapter’s connect method after letting the ReconnectionStrategy know.
Connection._create_channel(channel_number, on_open_callback)Create a new channel using the specified channel number and calling back the method specified byon_open_callback
Parameters
• channel_number (int) – The channel number to use
• on_open_callback (method) – The callback when the channel is opened
Connection._create_heartbeat_checker()Create a heartbeat checker instance if there is a heartbeat interval set.
Return type pika.heartbeat.Heartbeat
Connection._deliver_frame_to_channel(value)Deliver the frame to the channel specified in the frame.
Parameters value (pika.frame.Method) – The frame to deliver
Connection._detect_backpressure()Attempt to calculate if TCP backpressure is being applied due to our outbound buffer being larger than theaverage frame size over a window of frames.
Connection._ensure_closed()If the connection is not closed, close it.
Connection._flush_outbound()Adapters should override to flush the contents of outbound_buffer out along the socket.
Raises NotImplementedError
4.5. connection 91
pika Documentation, Release 0.9.9
Connection._get_body_frame_max_length()Calculate the maximum amount of bytes that can be in a body frame.
Return type int
Connection._get_credentials(method_frame)Get credentials for authentication.
Parameters method_frame (pika.frame.MethodFrame) – The Connection.Start frame
Return type tuple(str, str)
Connection._has_open_channelsReturns true if channels are open.
Return type bool
Connection._has_pending_callbacks(value)Return true if there are any callbacks pending for the specified frame.
Parameters value (pika.frame.Method) – The frame to check
Return type bool
Connection._init_connection_state()Initialize or reset all of the internal state variables for a given connection. On disconnect or reconnect allof the state needs to be wiped.
Connection._is_basic_deliver_frame(frame_value)Returns true if the frame is a Basic.Deliver
Parameters frame_value (pika.frame.Method) – The frame to check
Return type bool
Connection._is_connection_close_frame(value)Returns true if the frame is a Connection.Close frame.
Parameters value (pika.frame.Method) – The frame to check
Return type bool
Connection._is_method_frame(value)Returns true if the frame is a method frame.
Parameters value (pika.frame.Frame) – The frame to evaluate
Return type bool
Connection._is_protocol_header_frame(value)Returns True if it’s a protocol header frame.
Return type bool
Connection._next_channel_number()Return the next available channel number or raise on exception.
Return type int
Connection._on_channel_closeok(method_frame)Remove the channel from the dict of channels when Channel.CloseOk is sent.
Parameters method_frame (pika.frame.Method) – The response
Connection._on_close_ready()Called when the Connection is in a state that it can close after a close has been requested. This happens,for example, when all of the channels are closed that were open when the close request was made.
92 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Connection._on_connected()This is called by our connection Adapter to let us know that we’ve connected and we can notify ourconnection strategy.
Connection._on_connection_closed(method_frame, from_adapter=False)Called when the connection is closed remotely. The from_adapter value will be true if the connectionadapter has been disconnected from the broker and the method was invoked directly instead of by receivinga Connection.Close frame.
Parameters
• pika.frame.Method – The Connection.Close frame
• from_adapter (bool) – Called by the connection adapter
Connection._on_connection_open(method_frame)This is called once we have tuned the connection with the server and called the Connection.Open on theserver and it has replied with Connection.Ok.
Connection._on_connection_start(method_frame)This is called as a callback once we have received a Connection.Start from the server.
Parameters method_frame (pika.frame.Method) – The frame received
Raises UnexpectedFrameError
Connection._on_connection_tune(method_frame)Once the Broker sends back a Connection.Tune, we will set our tuning variables that have been returnedto us and kick off the Heartbeat monitor if required, send our TuneOk and then the Connection. Open rpccall on channel 0.
Parameters method_frame (pika.frame.Method) – The frame received
Connection._on_data_available(data_in)This is called by our Adapter, passing in the data from the socket. As long as we have buffer try and mapout frame data.
Parameters data_in (str) – The data that is available to read
Connection._process_callbacks(frame_value)Process the callbacks for the frame if the frame is a method frame and if it has any callbacks pending.
Parameters frame_value (pika.frame.Method) – The frame to process
Return type bool
Connection._process_connection_closed_callbacks()Process any callbacks that should be called when the connection is closed.
Connection._process_frame(frame_value)Process an inbound frame from the socket.
Parameters frame_value (pika.frame.Frame | pika.frame.Method) – The frame to process
Connection._read_frame()Try and read from the frame buffer and decode a frame.
Rtype tuple (int, pika.frame.Frame)
Connection._reject_out_of_band_delivery(channel_number, delivery_tag)Reject a delivery on the specified channel number and delivery tag because said channel no longer exists.
Parameters
• channel_number (int) – The channel number
4.5. connection 93
pika Documentation, Release 0.9.9
• delivery_tag (int) – The delivery tag
Connection._remove_callback(channel_number, method_frame)Remove the specified method_frame callback if it is set for the specified channel number.
Parameters
• channel_number (int) – The channel number to remove the callback on
• pika.object.Method – The method frame for the callback
Connection._remove_callbacks(channel_number, method_frames)Remove the callbacks for the specified channel number and list of method frames.
Parameters
• channel_number (int) – The channel number to remove the callback on
• method_frames (list) – The method frames for the callback
Connection._remove_connection_callbacks()Remove all callbacks for the connection
Connection._rpc(channel_number, method_frame, callback_method=None, accept-able_replies=None)
Make an RPC call for the given callback, channel number and method. acceptable_replies lists out whatresponses we’ll process from the server with the specified callback.
Parameters
• channel_number (int) – The channel number for the RPC call
• method_frame (pika.object.Method) – The method frame to call
• callback_method (method) – The callback for the RPC response
• acceptable_replies (list) – The replies this RPC call expects
Connection._send_connection_close(reply_code, reply_text)Send a Connection.Close method frame.
Parameters
• reply_code (int) – The reason for the close
• reply_text (str) – The text reason for the close
Connection._send_connection_open()Send a Connection.Open frame
Connection._send_connection_start_ok(authentication_type, response)Send a Connection.StartOk frame
Parameters
• authentication_type (str) – The auth type value
• response (str) – The encoded value to send
Connection._send_connection_tune_ok()Send a Connection.TuneOk frame
Connection._send_frame(frame_value)This appends the fully generated frame to send to the broker to the output buffer which will be then sentvia the connection adapter.
Parameters frame_value (pika.frame.Frame|pika.frame.ProtocolHeader) – The frame to write
94 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Connection._send_method(channel_number, method_frame, content=None)Constructs a RPC method frame and then sends it to the broker.
Parameters
• channel_number (int) – The channel number for the frame
• method_frame (pika.object.Method) – The method frame to send
• content (tuple) – If set, is a content frame, is tuple of properties and body.
Connection._set_connection_state(connection_state)Set the connection state.
Parameters connection_state (int) – The connection state to set
Connection._set_server_information(method_frame)Set the server properties and capabilities
Parameters method_frame (spec.connection.Start) – The Connection.Start frame
Connection._trim_frame_buffer(byte_count)Trim the leading N bytes off the frame buffer and increment the counter that keeps track of how manybytes have been read/used from the socket.
Parameters byte_count (int) – The number of bytes consumed
4.6 credentials
Credentials Classes
4.6.1 PlainCredentials
class pika.credentials.PlainCredentials(username, password, erase_on_connect=False)The PlainCredentials class returns the properly formatted username and password to the Connection. As ofthis version of Pika, only PlainCredentials are supported. To authenticate with Pika, simply create a credentialsobject passing in the username and password and pass that to the ConnectionParameters object.
If you do not pass in credentials to the ConnectionParameters object, it will create credentials for ‘guest’ withthe password of ‘guest’.
If you pass True to erase_on_connect the credentials will not be stored in memory after the Connection attempthas been made.
PlainCredentials.erase_credentials()Called by Connection when it no longer needs the credentials
PlainCredentials.response_for(start)Validate that this type of authentication is supported
Parameters start (spec.Connection.Start) – Connection.Start method
Return type tuple(str|None, str|None)
4.6.2 ExternalCredentials
class pika.credentials.ExternalCredentialsThe ExternalCredentials class allows the connection to use EXTERNAL authentication, generally with a clientSSL certificate.
4.6. credentials 95
pika Documentation, Release 0.9.9
ExternalCredentials.erase_credentials()Called by Connection when it no longer needs the credentials
ExternalCredentials.response_for(start)Validate that this type of authentication is supported
Parameters start (spec.Connection.Start) – Connection.Start method
Return type tuple(str or None, str or None)
4.7 data
The data module is for encoding and decoding of AMQP binary encoded data.
Note: This class is not intended for use by those using Pika in their applications. This documentation is for thosewho are extending Pika or otherwise working on the driver itself.
AMQP Table Encoding/Decoding
pika.data.encode_table(pieces, table)Encode a dict as an AMQP table appending the encded table to the pieces list passed in.
Parameters
• pieces (list) – Already encoded frame pieces
• table (dict) – The dict to encode
Return type int
pika.data.encode_value(pieces, value)Encode the value passed in and append it to the pieces list returning the the size of the encoded value.
Parameters
• pieces (list) – Already encoded values
• value (any) – The value to encode
Return type int
pika.data.decode_table(encoded, offset)Decode the AMQP table passed in from the encoded value returning the decoded result and the number of bytesread plus the offset.
Parameters
• encoded (str) – The binary encoded data to decode
• offset (int) – The starting byte offset
Return type tuple
pika.data.decode_value(encoded, offset)Decode the value passed in returning the decoded value and the number of bytes read in addition to the startingoffset.
Parameters
• encoded (str) – The binary encoded data to decode
• offset (int) – The starting byte offset
96 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Return type tuple
Raises pika.exceptions.InvalidFieldTypeException
4.8 exceptions
Pika specific exceptions
exception pika.exceptions.AMQPChannelError
exception pika.exceptions.AMQPConnectionError
exception pika.exceptions.AMQPError
exception pika.exceptions.AuthenticationError
exception pika.exceptions.BodyTooLongError
exception pika.exceptions.ChannelClosed
exception pika.exceptions.ChannelError
exception pika.exceptions.ConnectionClosed
exception pika.exceptions.ConsumerCancelled
exception pika.exceptions.DuplicateConsumerTag
exception pika.exceptions.IncompatibleProtocolError
exception pika.exceptions.InvalidChannelNumber
exception pika.exceptions.InvalidFieldTypeException
exception pika.exceptions.InvalidFrameError
exception pika.exceptions.InvalidMaximumFrameSize
exception pika.exceptions.InvalidMinimumFrameSize
exception pika.exceptions.MethodNotImplemented
exception pika.exceptions.NoFreeChannels
exception pika.exceptions.ProbableAccessDeniedError
exception pika.exceptions.ProbableAuthenticationError
exception pika.exceptions.ProtocolSyntaxError
exception pika.exceptions.ProtocolVersionMismatch
exception pika.exceptions.UnexpectedFrameError
exception pika.exceptions.UnspportedAMQPFieldException
4.9 frame
The frame module contains the object structure for mapping AMQP Classes and Methods on to Python objects. Inaddition frame contains the Dispatcher class which steps through the synchronous receipt order of frames received forBasic.Deliver and Basic.GetOk.
4.8. exceptions 97
pika Documentation, Release 0.9.9
Note: This class level documentation is not intended for use by those using Pika in their applications. This documen-tation is for those who are extending Pika or otherwise working on the driver itself.
Frame objects that do the frame demarshaling and marshaling.
4.9.1 Frame
class pika.frame.Frame(frame_type, channel_number)Base Frame object mapping. Defines a behavior for all child classes for assignment of core attributes andimplementation of the a core _marshal method which child classes use to create the binary AMQP frame.
_marshal(pieces)Create the full AMQP wire protocol frame data representation
Return type str
marshal()To be ended by child classes
:raises NotImplementedError
4.9.2 Method
class pika.frame.Method(channel_number, method)Base Method frame object mapping. AMQP method frames are mapped on top of this class for creating oraccessing their data and attributes.
marshal()Return the AMQP binary encoded value of the frame
Return type str
4.9.3 Header
class pika.frame.Header(channel_number, body_size, props)Header frame object mapping. AMQP content header frames are mapped on top of this class for creating oraccessing their data and attributes.
marshal()Return the AMQP binary encoded value of the frame
Return type str
4.9.4 Body
class pika.frame.Body(channel_number, fragment)Body frame object mapping class. AMQP content body frames are mapped on to this base class for get-ting/setting of attributes/data.
marshal()Return the AMQP binary encoded value of the frame
Return type str
98 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
4.9.5 ProtocolHeader
class pika.frame.ProtocolHeader(major=None, minor=None, revision=None)AMQP Protocol header frame class which provides a pythonic interface for creating AMQP Protocol headers
marshal()Return the full AMQP wire protocol frame data representation of the ProtocolHeader frame
Return type str
4.9.6 Frame Decoding
pika.frame.decode_frame(data_in)Receives raw socket data and attempts to turn it into a frame. Returns bytes used to make the frame and theframe
Parameters data_in (str) – The raw data stream
Return type tuple(bytes consumed, frame)
Raises pika.exceptions.InvalidFrameError
4.10 heartbeat
Note: This class is not intended for use by those using Pika in their applications. This documentation is for thosewho are extending Pika or otherwise working on the driver itself.
Handle AMQP Heartbeats
4.10.1 HeartbeatChecker
class pika.heartbeat.HeartbeatChecker(connection, interval, idle_count=2)Checks to make sure that our heartbeat is received at the expected intervals.
MAX_IDLE_COUNT = 2
_CONNECTION_FORCED = 320
_STALE_CONNECTION = ‘Too Many Missed Heartbeats, No reply in %i seconds’
activeReturn True if the connection’s heartbeat attribute is set to this instance.
:rtype True
bytes_received_on_connectionReturn the number of bytes received by the connection bytes object.
:rtype int
connection_is_idleReturns true if the byte count hasn’t changed in enough intervals to trip the max idle threshold.
received()Called when a heartbeat is received
4.10. heartbeat 99
pika Documentation, Release 0.9.9
send_and_check()Invoked by a timer to send a heartbeat when we need to, check to see if we’ve missed any heartbeats anddisconnect our connection if it’s been idle too long.
_close_connection()Close the connection with the AMQP Connection-Forced value.
_has_received_dataReturns True if the connection has received data on the connection.
Return type bool
_new_heartbeat_frame()Return a new heartbeat frame.
:rtype pika.frame.Heartbeat
_send_heartbeat_frame()Send a heartbeat frame on the connection.
_setup_timer()Use the connection objects delayed_call function which is implemented by the Adapter for calling thecheck_heartbeats function every interval seconds.
_start_timer()If the connection still has this object set for heartbeats, add a new timer.
_update_counters()Update the internal counters for bytes sent and received and the number of frames received
4.11 simplebuffer
Note: This class is not intended for use by those using Pika in their applications. This documentation is for thosewho are extending Pika or otherwise working on the driver itself.
4.11.1 SimpleBuffer
This is an implementation of a simple buffer. SimpleBuffer just handles a string of bytes. The clue, is that you canpop data from the beginning and append data to the end.
It’s ideal to use as a network buffer, from which you send data to the socket. Use this to avoid concatenating or splittinglarge strings.
class pika.simplebuffer.SimpleBuffer(data=None)A simple buffer that will handle storing, reading and sending strings to a socket.
>>> b = SimpleBuffer("abcdef")>>> b.read_and_consume(3)’abc’>>> b.write(None, ’’)>>> b.read(0)’’>>> repr(b)"<SimpleBuffer of 3 bytes, 6 total size, ’def’>">>> str(b)"<SimpleBuffer of 3 bytes, 6 total size, ’def’>"
100 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
>>> b.flush()>>> b.read(1)’’
SimpleBuffer.write(*data_strings)Append given strings to the buffer.
Parameters data_strings (str or unicode) – Value to write to the buffer
SimpleBuffer.read(size=None)Read the data from the buffer, at most ‘size’ bytes.
Parameters size (int) – The number of bytes to read
SimpleBuffer.consume(size)Move pointer and discard first ‘size’ bytes.
Parameters size (int) – The number of bytes to consume
SimpleBuffer.read_and_consume(size)Read up to ‘size’ bytes, also remove it from the buffer.
Parameters size (int) – The number of bytes to read and consume
Return type str
SimpleBuffer.send_to_socket(sd)Faster way of sending buffer data to socket ‘sd’.
Parameters sd (socket.socket) – The socket to send data to
Return type int
SimpleBuffer.flush()Remove all the data from buffer.
4.12 spec
Note: This class is not intended for use by those using Pika in their applications. This documentation is for thosewho are extending Pika or otherwise working on the driver itself.
spec.py is an auto-generated python module which is created by the codegen.py application. It contains all of theinformation about the protocol that is required for Pika to communicate with RabbitMQ.
class pika.spec.Connection
INDEX = 10
NAME = ‘Connection’
class Start(version_major=0, version_minor=9, server_properties=None, mechanisms=’PLAIN’, lo-cales=’en_US’)
INDEX = 655370
NAME = ‘Connection.Start’
synchronous
decode(encoded, offset=0)
4.12. spec 101
pika Documentation, Release 0.9.9
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Connection.StartOk(client_properties=None, mechanism=’PLAIN’, response=None, lo-cale=’en_US’)
INDEX = 655371
NAME = ‘Connection.StartOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Connection.Secure(challenge=None)
INDEX = 655380
NAME = ‘Connection.Secure’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
102 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Connection.SecureOk(response=None)
INDEX = 655381
NAME = ‘Connection.SecureOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Connection.Tune(channel_max=0, frame_max=0, heartbeat=0)
INDEX = 655390
NAME = ‘Connection.Tune’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Connection.TuneOk(channel_max=0, frame_max=0, heartbeat=0)
INDEX = 655391
4.12. spec 103
pika Documentation, Release 0.9.9
NAME = ‘Connection.TuneOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Connection.Open(virtual_host=’/’, capabilities=’‘, insist=False)
INDEX = 655400
NAME = ‘Connection.Open’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Connection.OpenOk(known_hosts=’‘)
INDEX = 655401
NAME = ‘Connection.OpenOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters
104 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Connection.Close(reply_code=None, reply_text=’‘, class_id=None, method_id=None)
INDEX = 655410
NAME = ‘Connection.Close’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Connection.CloseOk
INDEX = 655411
NAME = ‘Connection.CloseOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
4.12. spec 105
pika Documentation, Release 0.9.9
class pika.spec.Channel
INDEX = 20
NAME = ‘Channel’
class Open(out_of_band=’‘)
INDEX = 1310730
NAME = ‘Channel.Open’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Channel.OpenOk(channel_id=’‘)
INDEX = 1310731
NAME = ‘Channel.OpenOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Channel.Flow(active=None)
INDEX = 1310740
106 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
NAME = ‘Channel.Flow’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Channel.FlowOk(active=None)
INDEX = 1310741
NAME = ‘Channel.FlowOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Channel.Close(reply_code=None, reply_text=’‘, class_id=None, method_id=None)
INDEX = 1310760
NAME = ‘Channel.Close’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters
4.12. spec 107
pika Documentation, Release 0.9.9
• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Channel.CloseOk
INDEX = 1310761
NAME = ‘Channel.CloseOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class pika.spec.Access
INDEX = 30
NAME = ‘Access’
class Request(realm=’/data’, exclusive=False, passive=True, active=True, write=True, read=True)
INDEX = 1966090
NAME = ‘Access.Request’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
108 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Access.RequestOk(ticket=1)
INDEX = 1966091
NAME = ‘Access.RequestOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class pika.spec.Exchange
INDEX = 40
NAME = ‘Exchange’
class Declare(ticket=0, exchange=None, type=’direct’, passive=False, durable=False,auto_delete=False, internal=False, nowait=False, arguments={})
INDEX = 2621450
NAME = ‘Exchange.Declare’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
4.12. spec 109
pika Documentation, Release 0.9.9
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Exchange.DeclareOk
INDEX = 2621451
NAME = ‘Exchange.DeclareOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Exchange.Delete(ticket=0, exchange=None, if_unused=False, nowait=False)
INDEX = 2621460
NAME = ‘Exchange.Delete’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Exchange.DeleteOk
INDEX = 2621461
NAME = ‘Exchange.DeleteOk’
110 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Exchange.Bind(ticket=0, destination=None, source=None, routing_key=’‘, nowait=False, ar-guments={})
INDEX = 2621470
NAME = ‘Exchange.Bind’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Exchange.BindOk
INDEX = 2621471
NAME = ‘Exchange.BindOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties
4.12. spec 111
pika Documentation, Release 0.9.9
• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Exchange.Unbind(ticket=0, destination=None, source=None, routing_key=’‘, nowait=False,arguments={})
INDEX = 2621480
NAME = ‘Exchange.Unbind’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Exchange.UnbindOk
INDEX = 2621491
NAME = ‘Exchange.UnbindOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
112 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
class pika.spec.Queue
INDEX = 50
NAME = ‘Queue’
class Declare(ticket=0, queue=’‘, passive=False, durable=False, exclusive=False, auto_delete=False,nowait=False, arguments={})
INDEX = 3276810
NAME = ‘Queue.Declare’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Queue.DeclareOk(queue=None, message_count=None, consumer_count=None)
INDEX = 3276811
NAME = ‘Queue.DeclareOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Queue.Bind(ticket=0, queue=’‘, exchange=None, routing_key=’‘, nowait=False, argu-ments={})
4.12. spec 113
pika Documentation, Release 0.9.9
INDEX = 3276820
NAME = ‘Queue.Bind’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Queue.BindOk
INDEX = 3276821
NAME = ‘Queue.BindOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Queue.Purge(ticket=0, queue=’‘, nowait=False)
INDEX = 3276830
NAME = ‘Queue.Purge’
synchronous
decode(encoded, offset=0)
encode()
114 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Queue.PurgeOk(message_count=None)
INDEX = 3276831
NAME = ‘Queue.PurgeOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Queue.Delete(ticket=0, queue=’‘, if_unused=False, if_empty=False, nowait=False)
INDEX = 3276840
NAME = ‘Queue.Delete’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
4.12. spec 115
pika Documentation, Release 0.9.9
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Queue.DeleteOk(message_count=None)
INDEX = 3276841
NAME = ‘Queue.DeleteOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Queue.Unbind(ticket=0, queue=’‘, exchange=None, routing_key=’‘, arguments={})
INDEX = 3276850
NAME = ‘Queue.Unbind’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Queue.UnbindOk
INDEX = 3276851
NAME = ‘Queue.UnbindOk’
116 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class pika.spec.Basic
INDEX = 60
NAME = ‘Basic’
class Qos(prefetch_size=0, prefetch_count=0, global_=False)
INDEX = 3932170
NAME = ‘Basic.Qos’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.QosOk
INDEX = 3932171
NAME = ‘Basic.QosOk’
synchronous
decode(encoded, offset=0)
4.12. spec 117
pika Documentation, Release 0.9.9
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.Consume(ticket=0, queue=’‘, consumer_tag=’‘, no_local=False, no_ack=False, exclu-sive=False, nowait=False, arguments={})
INDEX = 3932180
NAME = ‘Basic.Consume’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.ConsumeOk(consumer_tag=None)
INDEX = 3932181
NAME = ‘Basic.ConsumeOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
118 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.Cancel(consumer_tag=None, nowait=False)
INDEX = 3932190
NAME = ‘Basic.Cancel’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.CancelOk(consumer_tag=None)
INDEX = 3932191
NAME = ‘Basic.CancelOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.Publish(ticket=0, exchange=’‘, routing_key=’‘, mandatory=False, immediate=False)
INDEX = 3932200
4.12. spec 119
pika Documentation, Release 0.9.9
NAME = ‘Basic.Publish’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.Return(reply_code=None, reply_text=’‘, exchange=None, routing_key=None)
INDEX = 3932210
NAME = ‘Basic.Return’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.Deliver(consumer_tag=None, delivery_tag=None, redelivered=False, ex-change=None, routing_key=None)
INDEX = 3932220
NAME = ‘Basic.Deliver’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters
120 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.Get(ticket=0, queue=’‘, no_ack=False)
INDEX = 3932230
NAME = ‘Basic.Get’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.GetOk(delivery_tag=None, redelivered=False, exchange=None, routing_key=None, mes-sage_count=None)
INDEX = 3932231
NAME = ‘Basic.GetOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
4.12. spec 121
pika Documentation, Release 0.9.9
class Basic.GetEmpty(cluster_id=’‘)
INDEX = 3932232
NAME = ‘Basic.GetEmpty’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.Ack(delivery_tag=0, multiple=False)
INDEX = 3932240
NAME = ‘Basic.Ack’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.Reject(delivery_tag=None, requeue=True)
INDEX = 3932250
NAME = ‘Basic.Reject’
synchronous
decode(encoded, offset=0)
122 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.RecoverAsync(requeue=False)
INDEX = 3932260
NAME = ‘Basic.RecoverAsync’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.Recover(requeue=False)
INDEX = 3932270
NAME = ‘Basic.Recover’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
4.12. spec 123
pika Documentation, Release 0.9.9
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.RecoverOk
INDEX = 3932271
NAME = ‘Basic.RecoverOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Basic.Nack(delivery_tag=0, multiple=False, requeue=True)
INDEX = 3932280
NAME = ‘Basic.Nack’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class pika.spec.Tx
INDEX = 90
124 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
NAME = ‘Tx’
class Select
INDEX = 5898250
NAME = ‘Tx.Select’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Tx.SelectOk
INDEX = 5898251
NAME = ‘Tx.SelectOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Tx.Commit
INDEX = 5898260
NAME = ‘Tx.Commit’
synchronous
4.12. spec 125
pika Documentation, Release 0.9.9
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Tx.CommitOk
INDEX = 5898261
NAME = ‘Tx.CommitOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Tx.Rollback
INDEX = 5898270
NAME = ‘Tx.Rollback’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
126 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Tx.RollbackOk
INDEX = 5898271
NAME = ‘Tx.RollbackOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class pika.spec.Confirm
INDEX = 85
NAME = ‘Confirm’
class Select(nowait=False)
INDEX = 5570570
NAME = ‘Confirm.Select’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
4.12. spec 127
pika Documentation, Release 0.9.9
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class Confirm.SelectOk
INDEX = 5570571
NAME = ‘Confirm.SelectOk’
synchronous
decode(encoded, offset=0)
encode()
_set_content(properties, body)If the method is a content frame, set the properties and body to be carried as attributes of the class.
Parameters• properties (pika.frame.Properties) – AMQP Basic Properties• body (str or unicode) – The message body
get_body()Return the message body if it is set.
Return type str|unicode
get_properties()Return the properties if they are set.
Return type pika.frame.Properties
class pika.spec.BasicProperties(content_type=None, content_encoding=None, headers=None,delivery_mode=None, priority=None, correlation_id=None,reply_to=None, expiration=None, message_id=None, times-tamp=None, type=None, user_id=None, app_id=None, clus-ter_id=None)
CLASSalias of Basic
INDEX = 60
NAME = ‘BasicProperties’
FLAG_CONTENT_TYPE = 32768
FLAG_CONTENT_ENCODING = 16384
FLAG_HEADERS = 8192
FLAG_DELIVERY_MODE = 4096
FLAG_PRIORITY = 2048
FLAG_CORRELATION_ID = 1024
FLAG_REPLY_TO = 512
FLAG_EXPIRATION = 256
FLAG_MESSAGE_ID = 128
FLAG_TIMESTAMP = 64
FLAG_TYPE = 32
128 Chapter 4. Pika Core Modules and Classes
pika Documentation, Release 0.9.9
FLAG_USER_ID = 16
FLAG_APP_ID = 8
FLAG_CLUSTER_ID = 4
decode(encoded, offset=0)
encode()
pika.spec.has_content(methodNumber)
4.13 simplebuffer
Note: This class is not intended for use by those using Pika in their applications. This documentation is for thosewho are extending Pika or otherwise working on the driver itself.
4.13.1 SimpleBuffer
Non-module specific functions shared by modules in the pika package
pika.utils.is_callable(handle)Returns a bool value if the handle passed in is a callable method/function
Parameters handle (any) – The object to check
Return type bool
4.13. simplebuffer 129
pika Documentation, Release 0.9.9
130 Chapter 4. Pika Core Modules and Classes
CHAPTER
FIVE
AUTHORS
• Tony Garnock-Jones
• Gavin M. Roy
131
pika Documentation, Release 0.9.9
132 Chapter 5. Authors
CHAPTER
SIX
CONTRIBUTORS
• Adam Flynn
• Ales Teska
• Alexey Myasnikov
• Anton V. Yanchenko
• Ask Solem
• Asko Soukka
• Brian K. Jones
• Charles Law
• David Strauss
• Fredrik Svensson
• George y
• Hunter Morris
• Jacek ‘Forger’ Całusinski
• Jan Urbanski
• Jason J. W. Williams
• Jonty Wareing
• Josh Braegger
• Josh Hansen
• Jozef Van Eenbergen
• Kamil Kisiel
• Kane
• Kyösti Herrala
• Lars van de Kerkhof
• Marek Majkowski
• Michael Kenney
• Michael Laing
• Milan Skuhra
133
pika Documentation, Release 0.9.9
• Njal Karevoll
• nonleaf
• Olivier Le Thanh Duong
• Pankrat
• Pavlobaron
• Peter Magnusson
• Raphaël De Giusti
• Roey Berman
• Samuel Stauffer
• Sigurd Høgsbro
• Tristan Penman
134 Chapter 6. Contributors
CHAPTER
SEVEN
INDICES AND TABLES
• genindex
• modindex
• search
135
pika Documentation, Release 0.9.9
136 Chapter 7. Indices and tables
PYTHON MODULE INDEX
ppika.adapters.asyncore_connection, ??pika.adapters.base_connection, ??pika.adapters.blocking_connection, ??pika.adapters.select_connection, ??pika.adapters.tornado_connection, ??pika.amqp_object, ??pika.callback, ??pika.channel, ??pika.connection, ??pika.credentials, ??pika.data, ??pika.exceptions, ??pika.frame, ??pika.heartbeat, ??pika.spec, ??
137