Type Annotations for boto3 SQS

1.42.3 · active · verified Sat Apr 11

This library provides PEP 561-compliant type annotations for the boto3 SQS service, generated by `mypy-boto3-builder`. It enhances development with static type checking for AWS SQS operations using boto3. The current version is 1.42.3, and it follows a frequent release cadence, typically aligning with boto3 updates and `mypy-boto3-builder` improvements.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to initialize an SQS client with `boto3` and use the provided type annotations (`SQSClient`, `SendMessageResultTypeDef`, `MessageTypeDef`) for `send_message` and `receive_message` operations. Type checkers like MyPy will use these annotations to validate your code.

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.")

view raw JSON →