IoT Platform：Gunakan SDK Platform IoT untuk Python

Prosedur

1. Inisialisasi SDK. Untuk informasi selengkapnya, lihat Inisialisasi SDK.

2. Buat permintaan API dengan membuat instans request yang nama kelasnya memiliki format ${nama API}+"Request".

3. Panggil metode "set"+${parameter permintaan} dari instans request untuk menentukan parameter permintaan.

4. Panggil metode ${nama API dengan kata-kata dihubungkan oleh garis bawah}(request) dari instans client untuk mendapatkan tanggapan. Metode yang namanya diakhiri dengan _async menunjukkan metode asinkron.

   Contoh: invoke_things_service(request) dan invoke_things_service_async(request).

Untuk informasi selengkapnya tentang operasi API Platform IoT, lihat Daftar operasi berdasarkan fungsi. Untuk informasi selengkapnya tentang parameter permintaan dan parameter respons setiap operasi API, lihat referensi API.

Contoh berikut menunjukkan cara memanggil operasi Pub untuk memublikasikan pesan ke topik. Untuk informasi selengkapnya tentang parameter permintaan, lihat Pub.

request = iot_models.PubRequest(
    # ID instans Platform IoT. 
    iot_instance_id='${iotInstanceId}',
    # KunciProduk produk. 
    product_key='${productKey}',
    # Isi pesan yang ingin Anda kirim. Enkode "hello world" dalam Base64 sebagai string. 
    message_content='aGVsbG8gd29ybGQ=',
    # Nama topik kustom yang digunakan oleh perangkat untuk menerima pesan. 
    topic_full_name='/${productKey}/${deviceName}/user/get',
    # Tingkat quality of service (QoS) pesan. Pesan QoS 0 dan QoS 1 didukung. 
    qos=0
)
response = client.pub(request)
print('response : ' + response)

Kode contoh lengkap

# -*- 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:
        """
        Gunakan ID AccessKey dan Rahasia AccessKey Anda untuk menginisialisasi klien.
        """
        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()
        # ID AccessKey Anda.
        config.access_key_id = os.environ.get('ALICLOUD_ACCESS_KEY_ID')
        # Rahasia AccessKey Anda.
        config.access_key_secret = os.environ.get('ALICLOUD_ACCESS_KEY_SECRET')
        # ID wilayah Anda.
        config.region_id = 'cn-shanghai'
        return Iot20180120Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        """
        Panggilan sinkron
        """
        try:
            client = Sample.create_client('${accessKey}', '${accessKeySecret}')
            request = iot_models.PubRequest(
                # ID instans Platform IoT. 
                iot_instance_id='${iotInstanceId}',
                # KunciProduk produk. 
                product_key='${productKey}',
                # Isi pesan yang ingin Anda kirim. Enkode "hello world" dalam Base64 sebagai string. 
                message_content='eyJ0ZXN0IjoidGFzayBwdWIgYnJvYWRjYXN0In0=',
                # Topik kustom yang digunakan untuk memublikasikan pesan. 
                topic_full_name='/${productKey}/${deviceName}/user/get',
                # Tingkat QoS pesan. Pesan QoS 0 dan QoS 1 didukung. 
                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:
        """
        Panggilan asinkron
        """
        try:
            client = Sample.create_client('${accessKey}', '${accessKeySecret}')
            request = iot_models.PubRequest(
                # ID instans Platform IoT. 
                iot_instance_id='${iotInstanceId}',
                # KunciProduk produk. 
                product_key='${productKey}',
                # Isi pesan yang ingin Anda kirim. Enkode "hello world" dalam Base64 sebagai string. 
                message_content='eyJ0ZXN0IjoidGFzayBwdWIgYnJvYWRjYXN0In0=',
                # Topik kustom yang digunakan untuk memublikasikan pesan. 
                topic_full_name='/${productKey}/${deviceName}/user/get',
                # Tingkat QoS pesan. Pesan QoS 0 dan QoS 1 didukung. 
                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:])