Python SDK Access Example

Development Environment

Python 3.0 or later is required. In this example, Python 3.9 is used.

Downloading the SDK

In this example, AMQP SDK python-qpid-proton (version 0.37.0) is used. You can run the following command to install the SDK of the latest version:

pip install python-qpid-proton

Sample Code

import threading
import time

from proton import SSLDomain
from proton.handlers import MessagingHandler
from proton.reactor import Container

# Reconnection times
reconnectTimes = 0


def current_time_millis():
    return str(int(round(time.time() * 1000)))


class AmqpClient(MessagingHandler):
    def __init__(self, host, port, accessKey, accessCode, queueName, instanceId):
        super(AmqpClient, self).__init__()
        self.host = host
        self.port = port
        self.accessKey = accessKey
        self.accessCode = accessCode
        self.queueName = queueName
        self.instanceId = instanceId

    def on_start(self, event):
        # Access domain name. For details, see "AMQP Client Access".
        url = "amqps://%s:%s" % (self.host, self.port)

        timestamp = current_time_millis()
        userName = "accessKey=" + self.accessKey + "|timestamp=" + timestamp + "|instanceId=" + self.instanceId
        passWord = self.accessCode
        # By default, the server certificate is not verified.
        sslDomain = SSLDomain(SSLDomain.MODE_CLIENT)
        sslDomain.set_peer_authentication(SSLDomain.ANONYMOUS_PEER)
        self.conn = event.container.connect(url, user=userName, password=passWord, heartbeat=60, ssl_domain=sslDomain,
                                            reconnect=False)
        event.container.create_receiver(self.conn, source=self.queueName)

    # Called when the connection is established.
    def on_connection_opened(self, event):
        global reconnectTimes
        reconnectTimes = 0
        print("Connection established, remoteUrl: %s", event.connection.hostname)

    # Called when the connection is cut off.
    def on_connection_closed(self, event):
        print("Connection closed: %s", self)
        ReconnectThread("reconnectThread").start()

    # Called when the remote end is disconnected due to an error.
    def on_connection_error(self, event):
        print("Connection error:%s", self)
        ReconnectThread("reconnectThread").start()

    # Called when an error occurs during AMQP connection establishment. Such errors include authentication and socket errors.
    def on_transport_error(self, event):
        if event.transport.condition:
            if event.transport.condition.info:
                print("%s: %s: %s" % (event.transport.condition.name, event.transport.condition.description,
                                      event.transport.condition.info))
            else:
                print("%s: %s" % (event.transport.condition.name, event.transport.condition.description))
        else:
            print("Unspecified transport error")
        ReconnectThread("reconnectThread").start()

    # Called when a message is received.
    def on_message(self, event):
        message = event.message
        content = message.body
        print("receive message: content=%s" % content)


class ReconnectThread(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name

    def run(self):
        global reconnectTimes
        reconnectTimes = reconnectTimes + 1
        time.sleep(15 if reconnectTimes > 15 else reconnectTimes)
        Container(AmqpClient(amqpHost, amqpPort, amqpAccessKey, amqpAccessCode, amqpQueueName, instanceId)).run()


# For details about how to set the following parameters, see Connection Configuration Parameters.
# AMQP access domain name
amqpHost = "127.0.0.1"

# AMQP access port
amqpPort = 5671

# Access key
amqpAccessKey = 'your AccessKey'

# Access code
amqpAccessCode = 'your AccessCode'

# Name of the subscription queue
amqpQueueName = 'DefaultQueue'

# Instance ID. This parameter is mandatory when multiple instances of the standard edition are purchased in the same region.
instanceId = ''

Container(AmqpClient(amqpHost, amqpPort, amqpAccessKey, amqpAccessCode, amqpQueueName, instanceId)).run()