Encapsulates AWS IoT actions.
| 18 | # snippet-start:[python.example_code.iot.IoTWrapper] |
| 19 | # snippet-start:[python.example_code.iot.IoTWrapper.decl] |
| 20 | class IoTWrapper: |
| 21 | """Encapsulates AWS IoT actions.""" |
| 22 | |
| 23 | def __init__(self, iot_client, iot_data_client=None): |
| 24 | """ |
| 25 | :param iot_client: A Boto3 AWS IoT client. |
| 26 | :param iot_data_client: A Boto3 AWS IoT Data Plane client. |
| 27 | """ |
| 28 | self.iot_client = iot_client |
| 29 | self.iot_data_client = iot_data_client |
| 30 | |
| 31 | @classmethod |
| 32 | def from_client(cls): |
| 33 | iot_client = boto3.client("iot") |
| 34 | iot_data_client = boto3.client("iot-data") |
| 35 | return cls(iot_client, iot_data_client) |
| 36 | # snippet-end:[python.example_code.iot.IoTWrapper.decl] |
| 37 | |
| 38 | |
| 39 | # snippet-start:[python.example_code.iot.CreateThing] |
| 40 | def create_thing(self, thing_name): |
| 41 | """ |
| 42 | Creates an AWS IoT thing. |
| 43 | |
| 44 | :param thing_name: The name of the thing to create. |
| 45 | :return: The name and ARN of the created thing. |
| 46 | """ |
| 47 | try: |
| 48 | response = self.iot_client.create_thing(thingName=thing_name) |
| 49 | logger.info("Created thing %s.", thing_name) |
| 50 | except ClientError as err: |
| 51 | if err.response["Error"]["Code"] == "ResourceAlreadyExistsException": |
| 52 | logger.info("Thing %s already exists. Skipping creation.", thing_name) |
| 53 | return None |
| 54 | logger.error( |
| 55 | "Couldn't create thing %s. Here's why: %s: %s", |
| 56 | thing_name, |
| 57 | err.response["Error"]["Code"], |
| 58 | err.response["Error"]["Message"], |
| 59 | ) |
| 60 | raise |
| 61 | else: |
| 62 | return response |
| 63 | |
| 64 | # snippet-end:[python.example_code.iot.CreateThing] |
| 65 | |
| 66 | # snippet-start:[python.example_code.iot.ListThings] |
| 67 | def list_things(self): |
| 68 | """ |
| 69 | Lists AWS IoT things. |
| 70 | |
| 71 | :return: The list of things. |
| 72 | """ |
| 73 | try: |
| 74 | things = [] |
| 75 | paginator = self.iot_client.get_paginator("list_things") |
| 76 | for page in paginator.paginate(): |
| 77 | things.extend(page["things"]) |
no outgoing calls