IoT Platform:Use IoT Platform SDK for Python

Procedure

1. Initialize the SDK. For more information, see Initialize the SDK.

2. Create an API request by generating a request instance whose class name is in the ${API name}+"Request" format.

3. Call the "set"+${Request parameter} method of the request instance to specify the request parameters.

4. Call the ${API name with words connected by underscores}(request) method of the client instance to obtain the response. A method whose name ends with _async indicates an asynchronous method.

   Examples: invoke_things_service(request) and invoke_things_service_async(request).

For more information about the API operations of IoT Platform, see List of operations by function. For more information about the request parameters and response parameters of each API operation, see the API reference.

The following example shows how to call the Pub operation to publish a message to a topic. For more information about request parameters, see Pub.

request = iot_models.PubRequest(
    # The ID of the IoT Platform instance. 
    iot_instance_id='${iotInstanceId}',
    # The ProductKey of the product. 
    product_key='${productKey}',
    # The message body that you want to send. Encode "hello world" in Base64 as a string. 
    message_content='aGVsbG8gd29ybGQ=',
    # The name of the custom topic that is used by a device to receive the message. 
    topic_full_name='/${productKey}/${deviceName}/user/get',
    # The quality of service (QoS) level of the message. QoS 0 and QoS 1 messages are supported. 
    qos=0
)
response = client.pub(request)
print('response : ' + response)

Complete sample code

# -*- coding: utf-8 -*-
import sys
from typing import List
from Tea.core import TeaCore
from alibabacloud_iot20180120.client import Client as IotClient
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_iot20180120 import models as iot_models
from alibabacloud_tea_console.client import Client as ConsoleClient
from alibabacloud_tea_util.client import Client as UtilClient

access_key_id = os.environ.get('ACCESS_KEY_ID')
access_key_secret = os.environ.get('ACCESS_KEY_SECRET')

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> IotClient:
        """
        Use your AccessKey ID and AccessKey secret to initialize the client.
        """
        Use your AccessKey ID and AccessKey secret to initialize the client.
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config()
        # Your AccessKey ID.
        config.access_key_id = os.environ.get('ALICLOUD_ACCESS_KEY_ID')
        # Your AccessKey secret.
        config.access_key_secret = os.environ.get('ALICLOUD_ACCESS_KEY_SECRET')
        # Your region ID.
        config.region_id = 'cn-shanghai'
        return Iot20180120Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        """
        Synchronous call
        """
        try:
            client = Sample.create_client('${accessKey}', '${accessKeySecret}')
            request = iot_models.PubRequest(
                # The ID of the IoT Platform instance. 
                iot_instance_id='${iotInstanceId}',
                # The ProductKey of the product. 
                product_key='${productKey}',
                # The message body that you want to send. Encode "hello world" in Base64 as a string. 
                message_content='eyJ0ZXN0IjoidGFzayBwdWIgYnJvYWRjYXN0In0=',
                # The custom topic that is used to publish the message. 
                topic_full_name='/${productKey}/${deviceName}/user/get',
                # The QoS level of the message. QoS 0 and QoS 1 messages are supported. 
                qos=0
            )
            response = client.pub(request)
            ConsoleClient.log(UtilClient.to_jsonstring(TeaCore.to_map(response)))
        except Exception as error:
            ConsoleClient.log(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        """
        Asynchronous call
        """
        try:
            client = Sample.create_client('${accessKey}', '${accessKeySecret}')
            request = iot_models.PubRequest(
                # The ID of the IoT Platform instance. 
                iot_instance_id='${iotInstanceId}',
                # The ProductKey of the product. 
                product_key='${productKey}',
                # The message body that you want to send. Encode "hello world" in Base64 as a string. 
                message_content='eyJ0ZXN0IjoidGFzayBwdWIgYnJvYWRjYXN0In0=',
                # The custom topic that is used to publish the message. 
                topic_full_name='/${productKey}/${deviceName}/user/get',
                # The QoS level of the message. QoS 0 and QoS 1 messages are supported. 
                qos=0
            )
            response = await client.pub_async(request)
            ConsoleClient.log(UtilClient.to_jsonstring(TeaCore.to_map(response)))
        except Exception as error:
            ConsoleClient.log(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])