Type Annotations for aiobotocore SQS

Warnings

• breaking Python 3.8 support was removed for all `mypy-boto3-builder` generated packages, including `types-aiobotocore-sqs`. Requires Python 3.9 or higher.
  Fix: Upgrade your Python environment to 3.9 or newer. Ensure your CI/CD pipelines reflect this change.
• gotcha `types-aiobotocore-sqs` is a stub-only library. It provides type annotations for `aiobotocore` but does not contain any runtime code. You *must* install `aiobotocore` separately for your application to function.
  Fix: Always install `aiobotocore` alongside `types-aiobotocore-sqs`, typically `pip install aiobotocore types-aiobotocore-sqs`.
• breaking The `mypy-boto3-builder` (version 8.9.0 and newer) introduced breaking changes in TypeDef naming conventions for some services. For example, `CreateDistributionRequestRequestTypeDef` became `CreateDistributionRequestTypeDef`. While this library targets `aiobotocore`, similar changes might occur for its generated TypeDefs.
  Fix: Review your imports of specific TypeDefs (`type_defs` module) and adjust names according to the latest package structure. Consult the changelog of `mypy-boto3-builder` for detailed changes.
• gotcha For optimal type checking, the version of `types-aiobotocore-sqs` should ideally match the major/minor version of your installed `aiobotocore` library to ensure type compatibility with the underlying API.
  Fix: Verify that your `types-aiobotocore-sqs` version is compatible with your `aiobotocore` version. The `types-aiobotocore-sqs` package typically specifies its `aiobotocore` dependency range.

Install

• pip install types-aiobotocore-sqs aiobotocore Install with aiobotocore

Imports

• SQSClient

  from types_aiobotocore_sqs.client import SQSClient

• SendMessageRequestRequestTypeDef

  from types_aiobotocore_sqs.type_defs import SendMessageRequestRequestTypeDef

• ReceiveMessageResponseTypeDef

  from types_aiobotocore_sqs.type_defs import ReceiveMessageResponseTypeDef

• SQSResource

  from types_aiobotocore_sqs.service_resource import SQSResource

  SQSResource is available via `service_resource`, not `client`.

Quickstart

This quickstart demonstrates how to use `types-aiobotocore-sqs` to add type hints to an `aiobotocore` SQS client. It covers creating a queue, sending a message, receiving and deleting a message, and finally cleaning up the queue. The `SQSClient` object and various TypeDefs are explicitly annotated for enhanced type checking.

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())