types-boto3-sqs

import boto3 import os from types_boto3_sqs import SQSClient from types_boto3_sqs.type_defs import SendMessageResultTypeDef, MessageTypeDef def get_sqs_client() -> SQSClient: # In a real application, AWS credentials and region would be configured # via environment variables, ~/.aws/credentials, or IAM roles. # Using placeholder for demonstration. session = boto3.Session( aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', 'DUMMY_KEY'), aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', 'DUMMY_SECRET'), region_name=os.environ.get('AWS_REGION', 'us-east-1') ) return session.client('sqs') def send_and_receive_message(queue_url: str) -> None: client: SQSClient = get_sqs_client() # Send a message send_response: SendMessageResultTypeDef = client.send_message( QueueUrl=queue_url, MessageBody='Hello SQS with types!' ) print(f"Message sent with ID: {send_response.get('MessageId')}") # Receive messages receive_response = client.receive_message( QueueUrl=queue_url, MaxNumberOfMessages=1, WaitTimeSeconds=5 ) messages: list[MessageTypeDef] = receive_response.get('Messages', []) if messages: for message in messages: print(f"Received message: {message.get('Body')}") # For production, remember to delete the message after processing # client.delete_message(QueueUrl=queue_url, ReceiptHandle=message['ReceiptHandle']) else: print("No messages received.") # Example usage (replace with your actual SQS queue URL) # queue_url = os.environ.get('SQS_QUEUE_URL', 'https://sqs.us-east-1.amazonaws.com/123456789012/my-test-queue') # if 'DUMMY' not in os.environ.get('AWS_ACCESS_KEY_ID', 'DUMMY_KEY'): # send_and_receive_message(queue_url) # else: # print("Skipping SQS operations: AWS credentials not configured. Set AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and SQS_QUEUE_URL.")