Python AWS SQS
SQS queues - sending, receiving and deleting messages and creating queues from Python services.
Zato lets you work with Amazon SQS queues directly from your Python services. You create an AWS connection in the Dashboard, and the SQS client is available under conn.sqs - the same client that boto3.client('sqs') returns, with credentials and configuration managed for you.
Sending messages
# -*- coding: utf-8 -*-
# stdlib
import json
# Zato
from zato.server.service import Service
class SendOrderEvent(Service):
input = 'order_id'
def handle(self):
# Get the connection by its Dashboard name
conn = self.aws['My AWS']
# Look up the queue ..
queue_url = conn.sqs.get_queue_url(QueueName='orders')['QueueUrl']
# .. and publish the message.
message = json.dumps({'order_id': self.request.input.order_id})
conn.sqs.send_message(QueueUrl=queue_url, MessageBody=message)
Receiving messages
Received messages stay invisible to other consumers for the queue's visibility timeout - delete each one after processing it, otherwise it will reappear.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class ProcessOrderEvents(Service):
def handle(self):
conn = self.aws['My AWS']
queue_url = conn.sqs.get_queue_url(QueueName='orders')['QueueUrl']
# Receive up to ten messages, waiting up to two seconds for them
response = conn.sqs.receive_message(
QueueUrl=queue_url,
MaxNumberOfMessages=10,
WaitTimeSeconds=2,
)
# There is no Messages key at all if the queue was empty
messages = response.get('Messages', [])
for message in messages:
# Process the message ..
self.logger.info('Processing order event: %s', message['Body'])
# .. and delete it so it is not delivered again.
conn.sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=message['ReceiptHandle'])
self.response.payload = {'processed': len(messages)}
Creating queues
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class CreateOrderQueue(Service):
def handle(self):
conn = self.aws['My AWS']
# Create the queue - the call also succeeds if the queue already exists
# with the same attributes, returning its URL.
response = conn.sqs.create_queue(
QueueName='orders',
Attributes={'VisibilityTimeout': '60'},
)
self.response.payload = {'queue_url': response['QueueUrl']}