Python SDK使用说明

调用步骤

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:])