AWS code-library documentation change
Summary
Added Python code examples for SNS operations including topic creation/deletion, message publishing with attributes, queue subscriptions with filters, and a complete cross-service scenario demonstrating SNS-SQS integration
Security assessment
The changes focus on adding functional examples for SNS operations without addressing security vulnerabilities. The IAM policy example demonstrates standard resource-based permissions but doesn't introduce new security features or address specific vulnerabilities.
Diff
diff --git a/code-library/latest/ug/python_3_sns_code_examples.md b/code-library/latest/ug/python_3_sns_code_examples.md index 39f88217f..e68699c2f 100644 --- a//code-library/latest/ug/python_3_sns_code_examples.md +++ b//code-library/latest/ug/python_3_sns_code_examples.md @@ -70,0 +71,66 @@ There's more on GitHub. Find the complete example and learn how to set up and ru + + class SnsWrapper: + """Wrapper class for managing Amazon SNS operations.""" + + def __init__(self, sns_client: Any) -> None: + """ + Initialize the SnsWrapper. + + :param sns_client: A Boto3 Amazon SNS client. + """ + self.sns_client = sns_client + + @classmethod + def from_client(cls) -> 'SnsWrapper': + """ + Create an SnsWrapper instance using a default boto3 client. + + :return: An instance of this class. + """ + sns_client = boto3.client('sns') + return cls(sns_client) + + + def create_topic( + self, + topic_name: str, + is_fifo: bool = False, + content_based_deduplication: bool = False + ) -> str: + """ + Create an SNS topic. + + :param topic_name: The name of the topic to create. + :param is_fifo: Whether to create a FIFO topic. + :param content_based_deduplication: Whether to use content-based deduplication for FIFO topics. + :return: The ARN of the created topic. + :raises ClientError: If the topic creation fails. + """ + try: + # Add .fifo suffix for FIFO topics + if is_fifo and not topic_name.endswith('.fifo'): + topic_name += '.fifo' + + attributes = {} + if is_fifo: + attributes['FifoTopic'] = 'true' + if content_based_deduplication: + attributes['ContentBasedDeduplication'] = 'true' + + response = self.sns_client.create_topic( + Name=topic_name, + Attributes=attributes + ) + + topic_arn = response['TopicArn'] + logger.info(f"Created topic: {topic_name} with ARN: {topic_arn}") + return topic_arn + + except ClientError as e: + error_code = e.response.get('Error', {}).get('Code', 'Unknown') + logger.error(f"Error creating topic {topic_name}: {error_code} - {e}") + raise + + + + @@ -110,0 +177,50 @@ There's more on GitHub. Find the complete example and learn how to set up and ru + + class SnsWrapper: + """Wrapper class for managing Amazon SNS operations.""" + + def __init__(self, sns_client: Any) -> None: + """ + Initialize the SnsWrapper. + + :param sns_client: A Boto3 Amazon SNS client. + """ + self.sns_client = sns_client + + @classmethod + def from_client(cls) -> 'SnsWrapper': + """ + Create an SnsWrapper instance using a default boto3 client. + + :return: An instance of this class. + """ + sns_client = boto3.client('sns') + return cls(sns_client) + + + def delete_topic(self, topic_arn: str) -> bool: + """ + Delete an SNS topic. + + :param topic_arn: The ARN of the topic to delete. + :return: True if successful. + :raises ClientError: If the topic deletion fails. + """ + try: + self.sns_client.delete_topic(TopicArn=topic_arn) + + logger.info(f"Deleted topic: {topic_arn}") + return True + + except ClientError as e: + error_code = e.response.get('Error', {}).get('Code', 'Unknown') + + if error_code == 'NotFound': + logger.warning(f"Topic not found: {topic_arn}") + return True # Already deleted + else: + logger.error(f"Error deleting topic: {error_code} - {e}") + raise + + + + @@ -317,0 +434,78 @@ Publish a message that takes different forms based on the protocol of the subscr + + class SnsWrapper: + """Wrapper class for managing Amazon SNS operations.""" + + def __init__(self, sns_client: Any) -> None: + """ + Initialize the SnsWrapper. + + :param sns_client: A Boto3 Amazon SNS client. + """ + self.sns_client = sns_client + + @classmethod + def from_client(cls) -> 'SnsWrapper': + """ + Create an SnsWrapper instance using a default boto3 client. + + :return: An instance of this class. + """ + sns_client = boto3.client('sns') + return cls(sns_client) + + + def publish_message( + self, + topic_arn: str, + message: str, + tone_attribute: Optional[str] = None, + deduplication_id: Optional[str] = None, + message_group_id: Optional[str] = None + ) -> str: + """ + Publish a message to an SNS topic. + + :param topic_arn: The ARN of the SNS topic. + :param message: The message content to publish. + :param tone_attribute: Optional tone attribute for message filtering. + :param deduplication_id: Optional deduplication ID for FIFO topics. + :param message_group_id: Optional message group ID for FIFO topics. + :return: The message ID of the published message. + :raises ClientError: If the message publication fails. + """ + try: + publish_args = { + 'TopicArn': topic_arn, + 'Message': message + } + + # Add message attributes if tone is specified + if tone_attribute: + publish_args['MessageAttributes'] = { + 'tone': { + 'DataType': 'String', + 'StringValue': tone_attribute + } + } + + # Add FIFO-specific parameters + if message_group_id: + publish_args['MessageGroupId'] = message_group_id + + if deduplication_id: + publish_args['MessageDeduplicationId'] = deduplication_id + + response = self.sns_client.publish(**publish_args) + + message_id = response['MessageId'] + logger.info(f"Published message to topic {topic_arn} with ID: {message_id}") + return message_id + + except ClientError as e: + error_code = e.response.get('Error', {}).get('Code', 'Unknown') + logger.error(f"Error publishing message to topic: {error_code} - {e}") + raise + + +