117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
import boto3
|
|
import base64
|
|
import math
|
|
|
|
|
|
class receiveFromSQS(object):
|
|
"""docstring for receiveFromSQS"""
|
|
|
|
def __init__(self, session, queueURL):
|
|
super(receiveFromSQS, self).__init__()
|
|
self.session = session
|
|
self.sqs = session.client('sqs')
|
|
self.queueURL = queueURL
|
|
|
|
@classmethod
|
|
def createSession(cls, profileName, queueURL):
|
|
session = boto3.Session(profile_name=profileName)
|
|
return cls(session, queueURL)
|
|
|
|
def getQueueLength(self):
|
|
attributeNames = ['ApproximateNumberOfMessages']
|
|
self.queueAttributes = self.sqs.get_queue_attributes(
|
|
QueueUrl=self.queueURL, AttributeNames=attributeNames
|
|
)
|
|
self.queueLength = int(
|
|
self.queueAttributes['Attributes']['ApproximateNumberOfMessages']
|
|
)
|
|
return self.queueLength
|
|
|
|
def _receiveSQSMessage(
|
|
self,
|
|
totalNumberOfMessages,
|
|
maxNumberOfMessages=10,
|
|
deleteOnReceipt=False,
|
|
):
|
|
self.response = []
|
|
loops = int(math.ceil(totalNumberOfMessages / maxNumberOfMessages))
|
|
for i in range(0, loops):
|
|
self.response.append(
|
|
self.sqs.receive_message(
|
|
QueueUrl=self.queueURL,
|
|
MaxNumberOfMessages=maxNumberOfMessages,
|
|
)
|
|
)
|
|
return self
|
|
|
|
def _extractMessageFromSQS(self, totalNumberOfMessages):
|
|
self.messages = []
|
|
try:
|
|
loops = len(self.response)
|
|
for i in range(0, loops):
|
|
if 'Messages' in self.response[i]:
|
|
_message = self.response[i]['Messages']
|
|
for rawMessage in _message:
|
|
self.messages.append(rawMessage['Body'])
|
|
else:
|
|
print("No messages in the queue")
|
|
except KeyError:
|
|
print("No messages in the queue")
|
|
return self
|
|
|
|
# def receiveMessage(
|
|
# self, b64=True, MaxNumberOfMessages=10, delete=False, endString='END'
|
|
# ):
|
|
# while True:
|
|
# response = self.sqs.receive_message(
|
|
# QueueUrl=self.queueURL,
|
|
# MaxNumberOfMessages=MaxNumberOfMessages
|
|
# )
|
|
# if 'Messages' in response:
|
|
# for message in response['Messages']:
|
|
# if b64:
|
|
# decoded = base64.b64decode(message['Body'])
|
|
# msg = decoded.decode()
|
|
# if msg == endString:
|
|
# return self
|
|
# else:
|
|
# msg = print(message['Body'])
|
|
# if msg == endString:
|
|
# return self
|
|
# else:
|
|
# print('Queue is empty')
|
|
# break
|
|
|
|
def receiveAllMessages(self, b64=True, deleteOnReceipt=False):
|
|
totalNumberOfMessages = self.getQueueLength()
|
|
self._receiveSQSMessage(
|
|
totalNumberOfMessages,
|
|
deleteOnReceipt=deleteOnReceipt
|
|
)
|
|
self._extractMessageFromSQS(
|
|
totalNumberOfMessages=totalNumberOfMessages
|
|
)
|
|
if b64:
|
|
pass # WRITE THIS IN!
|
|
return self
|
|
|
|
|
|
inst = receiveFromSQS.createSession(
|
|
profileName='plex-aws',
|
|
queueURL='https://sqs.eu-west-1.amazonaws'
|
|
'.com/745437999005/slack-bot.fifo',
|
|
)
|
|
|
|
inst.receiveAllMessages()
|
|
for item in inst.messages:
|
|
print(item)
|
|
|
|
# queueLength = inst.getQueueLength()
|
|
|
|
# response = inst._receiveSQSMessage(queueLength)._extractMessageFromSQS(
|
|
# queueLength
|
|
# )
|
|
# # messages = inst._receiveSQSMessage(12)._extractMessageFromSQS(12)
|
|
# for item in response.message:
|
|
# print(item)
|