AWS code-library documentation change
Summary
Added Python code example for creating IoT topic rules with SNS integration
Security assessment
The change demonstrates general IoT rule creation without security-specific content. No evidence of addressing security vulnerabilities.
Diff
diff --git a/code-library/latest/ug/iot_example_iot_CreateTopicRule_section.md b/code-library/latest/ug/iot_example_iot_CreateTopicRule_section.md index 72a09b547..e33bfb63c 100644 --- a//code-library/latest/ug/iot_example_iot_CreateTopicRule_section.md +++ b//code-library/latest/ug/iot_example_iot_CreateTopicRule_section.md @@ -304,0 +305,67 @@ There's more on GitHub. Find the complete example and learn how to set up and ru +Python + + +**SDK for Python (Boto3)** + + +###### Note + +There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iot#code-examples). + + + class IoTWrapper: + """Encapsulates AWS IoT actions.""" + + def __init__(self, iot_client, iot_data_client=None): + """ + :param iot_client: A Boto3 AWS IoT client. + :param iot_data_client: A Boto3 AWS IoT Data Plane client. + """ + self.iot_client = iot_client + self.iot_data_client = iot_data_client + + @classmethod + def from_client(cls): + iot_client = boto3.client("iot") + iot_data_client = boto3.client("iot-data") + return cls(iot_client, iot_data_client) + + def create_topic_rule(self, rule_name, topic, sns_action_arn, role_arn): + """ + Creates an AWS IoT topic rule. + + :param rule_name: The name of the rule. + :param topic: The MQTT topic to subscribe to. + :param sns_action_arn: The ARN of the SNS topic to publish to. + :param role_arn: The ARN of the IAM role. + """ + try: + self.iot_client.create_topic_rule( + ruleName=rule_name, + topicRulePayload={ + "sql": f"SELECT * FROM '{topic}'", + "actions": [ + {"sns": {"targetArn": sns_action_arn, "roleArn": role_arn}} + ], + }, + ) + logger.info("Created topic rule %s.", rule_name) + except ClientError as err: + if err.response["Error"]["Code"] == "ResourceAlreadyExistsException": + logger.info("Topic rule %s already exists. Skipping creation.", rule_name) + return + logger.error( + "Couldn't create topic rule. Here's why: %s: %s", + err.response["Error"]["Code"], + err.response["Error"]["Message"], + ) + raise + + + + + * For API details, see [CreateTopicRule](https://docs.aws.amazon.com/goto/boto3/iot-2015-05-28/CreateTopicRule) in _AWS SDK for Python (Boto3) API Reference_. + + + +