Type Annotations for boto3 SQS
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
- breaking Python 3.8 support was removed from `mypy-boto3-builder` (and consequently, from `types-boto3-*` packages) starting with version 8.12.0. Users on Python 3.8 should use older versions of `types-boto3-sqs` or upgrade their Python interpreter.
- breaking In `mypy-boto3-builder` version 8.9.0, there were breaking changes to TypeDef naming conventions. Specifically, TypeDefs for packed method arguments use shorter names where possible (e.g., `CreateDistributionRequestRequestTypeDef` -> `CreateDistributionRequestTypeDef`), and conflicting `Extra` postfixes were moved (e.g., `CreateDistributionExtraRequestTypeDef` -> `CreateDistributionRequestExtraTypeDef`). If you explicitly imported these TypeDefs, your code may break.
- gotcha The `types-boto3-sqs` package is specifically built for a corresponding `boto3` version (e.g., `types-boto3-sqs` 1.42.3 for `boto3` 1.42.3). While minor mismatches might work, significant version discrepancies between `boto3` and `types-boto3-sqs` can lead to inaccurate type hints or type-checking errors, as the underlying AWS API might have changed.
Install
-
pip install types-boto3-sqs
Imports
- SQSClient
from types_boto3_sqs import SQSClient
- MessageTypeDef
from types_boto3_sqs.type_defs import MessageTypeDef
Quickstart
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.")