Use the Python SDK

Steps

1. The SDK client initialization is complete. For more information, see SDK Initialization.

2. Create an API request object. This creates a request instance of the ${API_Name}Request class.

3. Use the request object instance and the set_${RequestParameterName} method to set the required request parameter values for the API.

4. Use the initialized client object to call the ${api_name_with_underscores}(request) method. This method returns the result of the API call. Methods with the _async suffix are asynchronous method calls.

   For example: invoke_things_service(request) and invoke_things_service_async(request).

For more information about IoT Platform cloud APIs, see API list. For descriptions of the request parameters in the request and the response parameters in the response, see the documentation for the specific API.

This topic uses the Pub API as an example to show how to publish a message to a topic. For 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 to send. A Base64-encoded string of "hello world".
    message_content='aGVsbG8gd29ybGQ=',
    # The custom topic of the device that will receive the message.
    topic_full_name='/${productKey}/${deviceName}/user/get',
    # The method to send the message. QoS 0 and QoS 1 are supported.
    qos=0
)
response = client.pub(request)
print('response : ' + response)

Complete code example

# -*- coding: utf-8 -*-
import os
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

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> IotClient:
        """
        Initialize the account client with an AccessKey ID and AccessKey secret.
        @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 IotClient(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        """
        Synchronous call method
        """
        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 to send. A Base64-encoded string of "hello world".
                message_content='eyJ0ZXN0IjoidGFzayBwdWIgYnJvYWRjYXN0In0=',
                # The custom topic of the device that will receive the message.
                topic_full_name='/${productKey}/${deviceName}/user/get',
                # The method to send the message. QoS 0 and QoS 1 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 method
        """
        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 to send. A Base64-encoded string of "hello world".
                message_content='eyJ0ZXN0IjoidGFzayBwdWIgYnJvYWRjYXN0In0=',
                # The custom topic of the device that will receive the message.
                topic_full_name='/${productKey}/${deviceName}/user/get',
                # The method to send the message. QoS 0 and QoS 1 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:])