Amazon SQS — boto3 in Python¶

import boto3

sqs = boto3.client("sqs", region_name="us-east-1")
print(sqs.meta.service_model.protocol)        # json

try:
    sqs.get_queue_url(QueueName="missing")
except sqs.exceptions.QueueDoesNotExist as e:
    print(e.response["Error"])
# {'Message': 'The specified queue does not exist.', 'Code': 'AWS.SimpleQueueService.NonExistentQueue',
#  'QueryErrorCode': 'QueueDoesNotExist', 'Type': 'Sender'}

std = sqs.create_queue(
    QueueName="orders",
    Attributes={
        "VisibilityTimeout": "60",
        "MessageRetentionPeriod": "86400",          # 1 day
        "ReceiveMessageWaitTimeSeconds": "20",      # long polling by default
    },
    tags={"team": "billing"},                       # note: lowercase "tags" in this call, "Tags" elsewhere
)["QueueUrl"]

fifo = sqs.create_queue(
    QueueName="orders.fifo",                        # the suffix is mandatory
    Attributes={
        "FifoQueue": "true",
        "ContentBasedDeduplication": "false",
        "DeduplicationScope": "messageGroup",       # with the next line: high-throughput FIFO
        "FifoThroughputLimit": "perMessageGroupId",
    },
)["QueueUrl"]
print(std)    # https://sqs.us-east-1.amazonaws.com/123456789012/orders      (moto's fake account)
print(fifo)   # https://sqs.us-east-1.amazonaws.com/123456789012/orders.fifo

url = sqs.get_queue_url(QueueName="orders")["QueueUrl"]
arn = sqs.get_queue_attributes(QueueUrl=url, AttributeNames=["QueueArn"])["Attributes"]["QueueArn"]
print(arn)    # arn:aws:sqs:us-east-1:123456789012:orders

sqs.tag_queue(QueueUrl=url, Tags={"env": "test", "owner": "qa"})
sqs.untag_queue(QueueUrl=url, TagKeys=["owner"])
print(sqs.list_queue_tags(QueueUrl=url)["Tags"])      # {'env': 'test', 'team': 'billing'}

import json

resp = sqs.send_message(
    QueueUrl=url,
    MessageBody=json.dumps({"order_id": "o-1", "amount": 9.5}),    # a string; at most 1 MiB
    MessageAttributes={
        "tenant":   {"DataType": "String", "StringValue": "acme"},
        "priority": {"DataType": "Number", "StringValue": "3"},
        "blob":     {"DataType": "Binary", "BinaryValue": b"\x00\x01"},
    },
    DelaySeconds=0,                                                 # per-message timer; not allowed on FIFO
)
print(sorted(k for k in resp if k != "ResponseMetadata"))
# ['MD5OfMessageAttributes', 'MD5OfMessageBody', 'MessageId']

resp = sqs.send_message(
    QueueUrl=fifo,
    MessageBody=json.dumps({"order_id": "o-1", "event": "created"}),
    MessageGroupId="customer-42",                    # required: ordering lane
    MessageDeduplicationId="order-o-1-created",      # required unless ContentBasedDeduplication is on
)
print(sorted(k for k in resp if k != "ResponseMetadata"))
# ['MD5OfMessageBody', 'MessageId', 'SequenceNumber']

resp = sqs.send_message_batch(QueueUrl=url, Entries=[
    {"Id": "0", "MessageBody": "a"},
    {"Id": "1", "MessageBody": "b", "DelaySeconds": 30},
])
resp["Successful"]   # [{'Id': '0', 'MessageId': '...', 'MD5OfMessageBody': '...'}, ...]
resp["Failed"]       # [{'Id': '1', 'SenderFault': False, 'Code': 'InternalError', 'Message': '...'}]

# app/producer.py
import json


class BatchSendError(Exception):
    def __init__(self, failed: list[dict]):
        super().__init__(f"{len(failed)} entries failed: {failed}")
        self.failed = failed


def send_batch(sqs, queue_url: str, bodies: list[dict], *, attempts: int = 3) -> list[str]:
    """Send any number of messages in batches of 10. Retries server-side failures, raises on sender faults."""
    message_ids: list[str] = []
    for start in range(0, len(bodies), 10):
        entries = [{"Id": str(i), "MessageBody": json.dumps(b)} for i, b in enumerate(bodies[start:start + 10])]
        for _ in range(attempts):
            resp = sqs.send_message_batch(QueueUrl=queue_url, Entries=entries)   # HTTP 200 even if entries failed
            message_ids += [ok["MessageId"] for ok in resp["Successful"]]
            failed = resp.get("Failed", [])
            if not failed:
                break
            if any(f["SenderFault"] for f in failed):                           # retrying cannot fix our own mistake
                raise BatchSendError(failed)
            retry_ids = {f["Id"] for f in failed}
            entries = [e for e in entries if e["Id"] in retry_ids]
        else:
            raise BatchSendError(failed)
    return message_ids

resp = sqs.receive_message(
    QueueUrl=url,
    MaxNumberOfMessages=10,                    # 1-10, default 1: always ask for 10 in a worker
    WaitTimeSeconds=20,                        # long polling; 0 = short polling
    VisibilityTimeout=120,                     # optional: overrides the queue default for these messages
    MessageAttributeNames=["tenant"],          # custom attributes are returned only when requested: names, "All", or "prefix.*"
    MessageSystemAttributeNames=["ApproximateReceiveCount", "SentTimestamp"],
)
for msg in resp.get("Messages", []):           # the key is absent, not an empty list, when nothing arrived
    print(msg["MessageId"], msg["Body"], msg["Attributes"], msg.get("MessageAttributes"))
    # ReceiptHandle is the only handle for DeleteMessage / ChangeMessageVisibility

['Attributes', 'Body', 'MD5OfBody', 'MD5OfMessageAttributes', 'MessageAttributes', 'MessageId', 'ReceiptHandle']
{'SentTimestamp': '1792586957156', 'ApproximateReceiveCount': '1'}
{'tenant': {'StringValue': 'acme', 'DataType': 'String'}, 'priority': {'StringValue': '3', 'DataType': 'Number'}}

sqs.delete_message(QueueUrl=url, ReceiptHandle=msg["ReceiptHandle"])

def delete_batch(sqs, queue_url: str, messages: list[dict]) -> list[dict]:
    """Delete received messages 10 at a time. Returns the entries that failed."""
    failed = []
    for i in range(0, len(messages), 10):
        entries = [{"Id": str(n), "ReceiptHandle": m["ReceiptHandle"]} for n, m in enumerate(messages[i:i + 10])]
        failed += sqs.delete_message_batch(QueueUrl=queue_url, Entries=entries).get("Failed", [])
    return failed                                          # a failed delete means the message will be redelivered

sqs.change_message_visibility(QueueUrl=url, ReceiptHandle=handle, VisibilityTimeout=120)   # 120 s from now
sqs.change_message_visibility(QueueUrl=url, ReceiptHandle=handle, VisibilityTimeout=0)     # give it back immediately
sqs.change_message_visibility_batch(QueueUrl=url, Entries=[
    {"Id": "0", "ReceiptHandle": handle, "VisibilityTimeout": 60},
])

def drain(sqs, queue_url: str):
    """Generator over received messages; stops at the first empty long poll."""
    while True:
        resp = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=10, WaitTimeSeconds=1)
        if not (messages := resp.get("Messages")):
            return
        yield from messages

sqs.purge_queue(QueueUrl=url)      # removes every message, including in-flight ones; takes up to 60 s
sqs.delete_queue(QueueUrl=url)     # the name is blocked for creation for 60 s

pages = sqs.get_paginator("list_queues").paginate(QueueNamePrefix="jobs-", PaginationConfig={"PageSize": 2})
urls = [u for page in pages for u in page.get("QueueUrls", [])]      # a page without queues has no "QueueUrls" key
# page sizes [2, 1] for three queues; ['jobs-0', 'jobs-1', 'jobs-2']