types-aiobotocore-sqs

import asyncio from aiobotocore.session import get_session from types_aiobotocore_sqs.client import SQSClient from types_aiobotocore_sqs.type_defs import ( SendMessageRequestRequestTypeDef, ReceiveMessageResponseTypeDef, CreateQueueResultTypeDef ) async def main(): session = get_session() async with session.create_client("sqs") as client: # Explicitly type hint the client for static analysis sqs_client: SQSClient = client queue_name = "my-test-queue" queue_url: str = "" # Create or get queue URL try: create_queue_response: CreateQueueResultTypeDef = await sqs_client.create_queue( QueueName=queue_name, Attributes={'DelaySeconds': '0', 'MessageRetentionPeriod': '345600'} ) queue_url = create_queue_response["QueueUrl"] print(f"Created queue: {queue_url}") except sqs_client.exceptions.QueueAlreadyExists: # type: ignore - exceptions are dynamic print(f"Queue '{queue_name}' already exists. Retrieving URL...") get_queue_response = await sqs_client.get_queue_url(QueueName=queue_name) queue_url = get_queue_response["QueueUrl"] # Send a message send_message_params: SendMessageRequestRequestTypeDef = { "QueueUrl": queue_url, "MessageBody": "Hello from types-aiobotocore-sqs!", "DelaySeconds": 0, } send_response = await sqs_client.send_message(**send_message_params) print("Message sent with ID:", send_response.get("MessageId")) # Receive messages receive_response: ReceiveMessageResponseTypeDef = await sqs_client.receive_message( QueueUrl=queue_url, MaxNumberOfMessages=1, WaitTimeSeconds=1 # Long-polling ) messages = receive_response.get("Messages", []) if messages: for message in messages: print(f"Received message: {message['Body']} (ID: {message['MessageId']})") # Delete the message after processing await sqs_client.delete_message( QueueUrl=queue_url, ReceiptHandle=message["ReceiptHandle"] ) print("Message deleted.") else: print("No messages received.") # Clean up: delete the queue await sqs_client.delete_queue(QueueUrl=queue_url) print(f"Deleted queue: {queue_url}") if __name__ == "__main__": # Ensure AWS credentials are configured (e.g., via ~/.aws/credentials or env vars) # This example requires access to create, send, receive, and delete SQS messages/queues. asyncio.run(main())