A small script to push messages from SQS to SNS

sqs_to_sns.py

# Usage
# $ python sqs_to_sns.py my-queue-name


import boto3
import sys
import queue
import threading
from datetime import datetime
import json
from uuid import uuid4

work_queue = queue.Queue()

sqs = boto3.resource('sqs')
sns = boto3.client('sns')
sqs_client = boto3.client('sqs')

from_q_name = sys.argv[1]
print(("From: " + from_q_name + " To: SNS"))

from_q = sqs.get_queue_by_name(QueueName=from_q_name)
to_q = sqs.get_queue_by_name(QueueName='backup-queue')

skipped = 0
processed = 0
total = 0


def process_queue():
    while True:
        messages = work_queue.get()

        global total
        total = len(messages)

        for message in messages:
            message_content = json.loads(message.body)

            message.delete()

            print("Backing up Message to dead letter queue - just in case. Id: " +
                  message_content['MessageId'])
            bodies = list()
            bodies.append({'Id': str(uuid4()), 'MessageBody': message.body})

            to_q.send_messages(Entries=bodies)

            response = sns.publish(
                TopicArn=message_content['TopicArn'], Message=message_content['Message'])
            print(("Published Message to Topic " + str(message_content['MessageId']) +
                   ". To TopicArn: " + message_content['TopicArn'] + ". Received response " + json.dumps(response)))

            global processed
            processed = processed + 1


for i in range(10):
    t = threading.Thread(target=process_queue)
    t.daemon = True
    t.start()

while True:
    messages = list()
    for message in from_q.receive_messages(
            MaxNumberOfMessages=10,
            VisibilityTimeout=123,
            WaitTimeSeconds=20):
        messages.append(message)
    work_queue.put(messages)

work_queue.join()
print("Processed " + str(processed) + " of " + str(total) +
      ". Skipped " + str(skipped) + " messages. Exiting")