調用步驟

1. 已完成SDK客戶端初始化。具體內容，請參見初始化SDK。

2. 創建對應API的調用請求，生成${API名稱}+"Request"類對象實例request。

3. 通過對象實例request，使用"set_"+${請求參數的名稱}方法，設置API所需的請求參數值。

4. 通過初始化SDK的對象實例client，使用${經過下劃線分隔單詞的API名稱}(request)方法，承載API調用結果。其中，后綴為_async的方法表示異步方法。

   例如：invoke_things_service(request)，invoke_things_service_async(request)。

物聯網平臺云端API，請參見API列表。有關request中請求參數說明，及response中返回參數說明，請參見對應API文檔。

本文以調用Pub接口發布消息到Topic為例。請求參數信息，請參見Pub。

request = iot_models.PubRequest(
    # 物聯網平臺實例ID。
    iot_instance_id='${iotInstanceId}',
    # 產品ProductKey。
    product_key='${productKey}',
    # 要發送的消息主體，hello world Base64 String。
    message_content='aGVsbG8gd29ybGQ=',
    # 要接收消息的設備的自定義Topic。
    topic_full_name='/${productKey}/${deviceName}/user/get',
    # 指定消息的發送方式，支持QoS0和QoS1。
    qos=0
)
response = client.pub(request)
print('response : ' + response)

完整示例代碼

# -*- 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:
        """
        使用AK&SK初始化Client
        """
        使用AK&SK初始化賬號Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config()
        # 您的AccessKey ID
        config.access_key_id = os.environ.get('ALICLOUD_ACCESS_KEY_ID')
        # 您的AccessKey Secret
        config.access_key_secret = os.environ.get('ALICLOUD_ACCESS_KEY_SECRET')
        # 您的可用區ID
        config.region_id = 'cn-shanghai'
        return Iot20180120Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        """
        同步調用方式
        """
        try:
            client = Sample.create_client('${accessKey}', '${accessKeySecret}')
            request = iot_models.PubRequest(
                # 物聯網平臺實例ID。
                iot_instance_id='${iotInstanceId}',
                # 產品ProductKey。
                product_key='${productKey}',
                # 要發送的消息主體，hello world Base64 String。
                message_content='eyJ0ZXN0IjoidGFzayBwdWIgYnJvYWRjYXN0In0=',
                # 要接收消息的設備的自定義Topic。
                topic_full_name='/${productKey}/${deviceName}/user/get',
                # 指定消息的發送方式，支持QoS0和QoS1。
                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:
        """
        異步調用方式
        """
        try:
            client = Sample.create_client('${accessKey}', '${accessKeySecret}')
            request = iot_models.PubRequest(
                # 物聯網平臺實例ID。
                iot_instance_id='${iotInstanceId}',
                # 產品ProductKey。
                product_key='${productKey}',
                # 要發送的消息主體，hello world Base64 String。
                message_content='eyJ0ZXN0IjoidGFzayBwdWIgYnJvYWRjYXN0In0=',
                # 要接收消息的設備的自定義Topic。
                topic_full_name='/${productKey}/${deviceName}/user/get',
                # 指定消息的發送方式，支持QoS0和QoS1。
                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:])