IoT Platform:IoT Platform SDK for Python の使用

手順

1. SDK を初期化します。詳細については、「SDK の初期化」をご参照ください。

2. リクエスト インスタンスを生成することで API リクエストを作成します。インスタンスのクラス名は ${API 名}+"Request" フォーマットです。

3. "set"+${リクエスト パラメーター} メソッドを リクエスト インスタンスで呼び出して、リクエスト パラメーターを指定します。

4. ${アンダースコアで接続された単語を含む API 名}(リクエスト)client インスタンスの ${単語がアンダースコアで接続された API 名}(request) メソッドを呼び出して、レスポンスを取得します。名前が . で終わるメソッドは、非同期メソッドを示します

   例: invoke_things_service(request) および invoke_things_service_async(request)。

IoT Platform の API 操作の詳細については、「関数別の操作リスト」をご参照ください。各 API 操作のリクエストパラメーターとレスポンスパラメーターの詳細については、API リファレンスをご参照ください。

次の例は、Pub 操作を呼び出して、トピックにメッセージをパブリッシュする方法を示しています。リクエストパラメーターの詳細については、「Pub」をご参照ください。

request = iot_models.PubRequest(
    # IoT Platform インスタンスの ID。
    iot_instance_id='${iotInstanceId}',
    # プロダクトの ProductKey。
    product_key='${productKey}',
    # 送信するメッセージ本文。"hello world" を Base64 で文字列としてエンコードします。
    message_content='aGVsbG8gd29ybGQ=',
    # デバイスがメッセージを受信するために使用するカスタムトピックの名前。
    topic_full_name='/${productKey}/${deviceName}/user/get',
    # メッセージのサービス品質 (QoS) レベル。QoS 0 および QoS 1 メッセージがサポートされています。
    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:
        """
        アクセスキー ID とアクセスキーシークレットを使用してクライアントを初期化します。
        """
        # アクセスキー ID とアクセスキーシークレットを使用してクライアントを初期化します。
        # @param access_key_id:
        # @param access_key_secret:
        # @return: Client
        # @throws Exception
        config = open_api_models.Config()
        # アクセスキー ID。
        config.access_key_id = os.environ.get('ALICLOUD_ACCESS_KEY_ID')
        # アクセスキーシークレット。
        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(
                # IoT Platform インスタンスの ID。
                iot_instance_id='${iotInstanceId}',
                # プロダクトの ProductKey。
                product_key='${productKey}',
                # 送信するメッセージ本文。"hello world" を Base64 で文字列としてエンコードします。
                message_content='eyJ0ZXN0IjoidGFzayBwdWIgYnJvYWRjYXN0In0=',
                # メッセージをパブリッシュするために使用するカスタムトピック。
                topic_full_name='/${productKey}/${deviceName}/user/get',
                # メッセージの QoS レベル。QoS 0 および QoS 1 メッセージがサポートされています。
                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(
                # IoT Platform インスタンスの ID。
                iot_instance_id='${iotInstanceId}',
                # プロダクトの ProductKey。
                product_key='${productKey}',
                # 送信するメッセージ本文。"hello world" を Base64 で文字列としてエンコードします。
                message_content='eyJ0ZXN0IjoidGFzayBwdWIgYnJvYWRjYXN0In0=',
                # メッセージをパブリッシュするために使用するカスタムトピック。
                topic_full_name='/${productKey}/${deviceName}/user/get',
                # メッセージの QoS レベル。QoS 0 および QoS 1 メッセージがサポートされています。
                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:])