Updated on 2023-10-18 GMT+08:00

Python SDK Access Example

This topic describes how to use a Python 3 SDK to connect to the Huawei Cloud IoT platform and receive subscribed messages from the platform based on AMQP.

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

You can also manually install it by referring to Installing 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()