TL;DR: This Python script lets you easily debug AWS SQS queues locally via LocalStack. Features include listing messages, purging queues, and selecting from available queues interactively — all without touching the AWS cloud.
When working with Amazon SQS during development, setting up a fast feedback loop can be a challenge — especially if you want to avoid pushing messages to a real AWS account. That’s where LocalStack comes in, and this little utility script can make your life easier.
Here’s a breakdown of what it does and how to use it.
What This Script Does
- Connects to a local SQS endpoint (e.g., LocalStack).
- Lists all queues and lets you interactively choose one.
- Displays message details with optional JSON body parsing.
- Purges all messages in a queue (with confirmation).
- Simple CLI interface for selecting actions.
Requirements
Python 3.12+
boto3
Docker
LocalStack 4.4 : installation
awslocal : https://github.com/localstack/awscli-local
Localstack setup
1. Run LocalStack on Docker :
docker run \
--rm -d \
-p 127.0.0.1:4566:4566 \
-p 127.0.0.1:4510-4559:4510-4559 \
-v /var/run/docker.sock:/var/run/docker.sock \
-e DEBUG=1 -e SQS_ENDPOINT_STRATEGY=domain \
localstack/localstack
2. Create FIFO SQS queues
awslocal sqs create-queue \
--queue-name monitoring-queue.fifo \
--attributes "FifoQueue=true" \
--region eu-west-1
awslocal sqs create-queue \
--queue-name monitoring-queue-dlq.fifo \
--attributes "FifoQueue=true" \
--region eu-west-1
3. Send a message to FIFO SQS
awslocal sqs send-message \
--queue-url http://sqs.eu-west-1.localhost.localstack.cloud:4566/000000000000/monitoring-queue.fifo \
--message-body '{"status":"delivered"}' \
--message-group-id message-group \
--message-deduplication-id 1
Configuration
Don't forget to install boto3 via pip
The script reads a few environment variables:
REGION = os.getenv("REGION", "eu-west-1")
SQS_ENDPOINT = os.getenv("SQS_ENDPOINT", "https://localhost.localstack.cloud:4566")
You're free to adjust these to match your LocalStack setup.
It uses boto3 and assumes test credentials for local use:
sqs = boto3.client(
"sqs",
region_name=REGION,
endpoint_url=SQS_ENDPOINT,
aws_access_key_id="test",
aws_secret_access_key="test"
)
Features Breakdown
1. List Messages in a Queue
- Prints up to 10 messages per call, with rich metadata:
- Sent timestamp (human-readable)
- Approximate receive count
- Message group and deduplication IDs
- Sequence number (for FIFO queues)
- Parsed JSON body (if double-encoded)
def list_messages():
print("Checking messages in queue...\n")
selected_queue_url = list_queues_and_choose()
response = sqs.receive_message(
QueueUrl=selected_queue_url,
AttributeNames=['All'],
MessageAttributeNames=['All'],
MaxNumberOfMessages=10,
WaitTimeSeconds=2,
VisibilityTimeout=0
)
messages = response.get("Messages", [])
if not messages:
print("No messages available in the queue.")
return
for idx, msg in enumerate(messages, 1):
print(f"\n--- Message #{idx} ---")
print(f"MessageId: {msg['MessageId']}")
print(f"ReceiptHandle: {msg['ReceiptHandle']}")
# SQS attributes
attrs = msg.get("Attributes", {})
print(f"SentTimestamp: {human_time(attrs.get('SentTimestamp', '0'))}")
print(f"ReceiveCount: {attrs.get('ApproximateReceiveCount')}")
print(f"MessageGroupId: {attrs.get('MessageGroupId')}")
print(f"MessageDeduplicationId: {attrs.get('MessageDeduplicationId')}")
print(f"SequenceNumber: {attrs.get('SequenceNumber')}")
print(f"VisibilityTimeout Active (IsVisible): {'false' if int(attrs.get('ApproximateReceiveCount', 0)) > 0 else 'true'}")
# Raw Body
body = msg["Body"]
print(f"\nRaw Body:\n{body}")
2. Purge Queue
- Deletes all messages from the selected queue after confirmation. Useful for clearing test data.
def purge_queue():
print("Which queue do you want to purge: \n")
selected_queue_url = list_queues_and_choose()
if selected_queue_url:
confirm = input("Are you sure you want to purge the queue? This deletes ALL messages. (yes/no): ")
if confirm.lower() == "yes":
sqs.purge_queue(QueueUrl=selected_queue_url)
print("Queue purged.")
else:
print("Purge cancelled.")
3. List All Queues
- Simply prints all available queues registered with your local SQS.
def list_queues():
list = sqs.list_queues()
print(json.dumps(list, indent=4))
Why This Is Useful
- Faster feedback loop with local development.
- No AWS charges or IAM concerns.
- Understand SQS internals and debug message structures.
- Great for simulating dead-letter queue behavior.
Final Thoughts
If you’re working with SQS and want to test locally with minimal overhead, this script will save you time and make debugging much more transparent.
Github Source : https://github.com/axeldlv/aws-sqs-debugging
Let me know in the comments how you’d improve it or if you want a follow-up post to add more features!
Top comments (0)