AWS code-library documentation change
Summary
Added SqsWrapper class with receive_messages method for message retrieval
Security assessment
Basic message receiving implementation. No security vulnerabilities or security features addressed.
Diff
diff --git a/code-library/latest/ug/python_3_sqs_code_examples.md b/code-library/latest/ug/python_3_sqs_code_examples.md index 9dc01381b..9e3539777 100644 --- a//code-library/latest/ug/python_3_sqs_code_examples.md +++ b//code-library/latest/ug/python_3_sqs_code_examples.md @@ -67,0 +68,58 @@ There's more on GitHub. Find the complete example and learn how to set up and ru + + class SqsWrapper: + """Wrapper class for managing Amazon SQS operations.""" + + def __init__(self, sqs_client: Any) -> None: + """ + Initialize the SqsWrapper. + + :param sqs_client: A Boto3 Amazon SQS client. + """ + self.sqs_client = sqs_client + + @classmethod + def from_client(cls) -> 'SqsWrapper': + """ + Create an SqsWrapper instance using a default boto3 client. + + :return: An instance of this class. + """ + sqs_client = boto3.client('sqs') + return cls(sqs_client) + + + def create_queue(self, queue_name: str, is_fifo: bool = False) -> str: + """ + Create an SQS queue. + + :param queue_name: The name of the queue to create. + :param is_fifo: Whether to create a FIFO queue. + :return: The URL of the created queue. + :raises ClientError: If the queue creation fails. + """ + try: + # Add .fifo suffix for FIFO queues + if is_fifo and not queue_name.endswith('.fifo'): + queue_name += '.fifo' + + attributes = {} + if is_fifo: + attributes['FifoQueue'] = 'true' + + response = self.sqs_client.create_queue( + QueueName=queue_name, + Attributes=attributes + ) + + queue_url = response['QueueUrl'] + logger.info(f"Created queue: {queue_name} with URL: {queue_url}") + return queue_url + + except ClientError as e: + error_code = e.response.get('Error', {}).get('Code', 'Unknown') + logger.error(f"Error creating queue {queue_name}: {error_code} - {e}") + raise + + + + @@ -149,0 +208,70 @@ There's more on GitHub. Find the complete example and learn how to set up and ru + + class SqsWrapper: + """Wrapper class for managing Amazon SQS operations.""" + + def __init__(self, sqs_client: Any) -> None: + """ + Initialize the SqsWrapper. + + :param sqs_client: A Boto3 Amazon SQS client. + """ + self.sqs_client = sqs_client + + @classmethod + def from_client(cls) -> 'SqsWrapper': + """ + Create an SqsWrapper instance using a default boto3 client. + + :return: An instance of this class. + """ + sqs_client = boto3.client('sqs') + return cls(sqs_client) + + + def delete_messages(self, queue_url: str, messages: List[Dict[str, Any]]) -> bool: + """ + Delete messages from an SQS queue in batches. + + :param queue_url: The URL of the queue. + :param messages: List of messages to delete. + :return: True if successful. + :raises ClientError: If deleting messages fails. + """ + try: + if not messages: + return True + + # Build delete entries for batch delete + delete_entries = [] + for i, message in enumerate(messages): + delete_entries.append({ + 'Id': str(i), + 'ReceiptHandle': message['ReceiptHandle'] + }) + + # Delete messages in batches of 10 (SQS limit) + batch_size = 10 + for i in range(0, len(delete_entries), batch_size): + batch = delete_entries[i:i + batch_size] + + response = self.sqs_client.delete_message_batch( + QueueUrl=queue_url, + Entries=batch + ) + + # Check for failures + if 'Failed' in response and response['Failed']: + for failed in response['Failed']: + logger.warning(f"Failed to delete message: {failed}") + + logger.info(f"Deleted {len(messages)} messages from {queue_url}") + return True + + except ClientError as e: + error_code = e.response.get('Error', {}).get('Code', 'Unknown') + logger.error(f"Error deleting messages: {error_code} - {e}") + raise + + + + @@ -183,0 +312,50 @@ There's more on GitHub. Find the complete example and learn how to set up and ru + + class SqsWrapper: + """Wrapper class for managing Amazon SQS operations.""" + + def __init__(self, sqs_client: Any) -> None: + """ + Initialize the SqsWrapper. + + :param sqs_client: A Boto3 Amazon SQS client. + """ + self.sqs_client = sqs_client + + @classmethod + def from_client(cls) -> 'SqsWrapper': + """ + Create an SqsWrapper instance using a default boto3 client. + + :return: An instance of this class. + """ + sqs_client = boto3.client('sqs') + return cls(sqs_client) + + + def delete_queue(self, queue_url: str) -> bool: + """ + Delete an SQS queue. + + :param queue_url: The URL of the queue to delete. + :return: True if successful. + :raises ClientError: If the queue deletion fails. + """ + try: + self.sqs_client.delete_queue(QueueUrl=queue_url) + + logger.info(f"Deleted queue: {queue_url}") + return True + + except ClientError as e: + error_code = e.response.get('Error', {}).get('Code', 'Unknown') + + if error_code == 'AWS.SimpleQueueService.NonExistentQueue': + logger.warning(f"Queue not found: {queue_url}") + return True # Already deleted + else: + logger.error(f"Error deleting queue: {error_code} - {e}") + raise + + + + @@ -188,0 +367,63 @@ There's more on GitHub. Find the complete example and learn how to set up and ru +The following code example shows how to use `GetQueueAttributes`. + +**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/cross_service/topics_and_queues#code-examples). + + + class SqsWrapper: + """Wrapper class for managing Amazon SQS operations.""" + + def __init__(self, sqs_client: Any) -> None: