fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-01 02:08:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

Converted SQS broker to Boto3

Browse Source
This commit is contained in:
Ilan Steemers
2015-09-10 11:08:33 +02:00
parent 0fe6c4cbde
commit 0b9df26218
3 changed files with 23 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/brokers/aws_sqs.py
+16 -23
View File
@@ -1,19 +1,17 @@
from django_q.conf import Conf
from django_q.brokers import Broker
import boto.sqs
from boto.sqs.message import RawMessage
from boto3 import Session
class Sqs(Broker):
def __init__(self, list_key=Conf.PREFIX):
self.sqs = None
super(Sqs, self).__init__(list_key)
self.queue = self.get_queue()
def enqueue(self, task):
m = RawMessage()
m.set_body(task)
self.queue.write(m)
return m.id
response = self.queue.send_message(MessageBody=task)
return response.get('MessageId')
def dequeue(self):
# sqs supports max 10 messages in bulk
@@ -23,50 +21,45 @@ class Sqs(Broker):
if len(self.task_cache) > 0:
t = self.task_cache.pop()
else:
tasks = self.queue.get_messages(num_messages=Conf.BULK, visibility_timeout=Conf.RETRY)
tasks = self.queue.receive_messages(MaxNumberOfMessages=Conf.BULK, VisibilityTimeout=Conf.RETRY)
if tasks:
t = tasks.pop()
if tasks:
self.task_cache = tasks
if t:
return t.receipt_handle, t.get_body()
return t.receipt_handle, t.body
def acknowledge(self, task_id):
return self.delete(task_id)
def queue_size(self):
return self.queue.count()
return int(self.queue.attributes['ApproximateNumberOfMessages'])
def delete(self, task_id):
m = RawMessage()
m.receipt_handle = task_id
return self.queue.delete_message(m)
message = self.sqs.Message(self.queue.url, task_id)
message.delete()
def fail(self, task_id):
self.delete(task_id)
def delete_queue(self):
self.connection.delete_queue(self.queue)
self.queue.delete()
def purge_queue(self):
self.queue.purge()
def ping(self):
try:
self.connection.get_all_queues()
return True
except Exception as e:
raise e
return 'sqs' in self.connection.get_available_resources()
def info(self):
return 'AWS SQS'
@staticmethod
def get_connection(list_key=Conf.PREFIX):
conn = boto.sqs.connect_to_region(Conf.SQS['aws_region'],
aws_access_key_id=Conf.SQS['aws_access_key_id'],
aws_secret_access_key=Conf.SQS['aws_secret_access_key'])
return conn
return Session(aws_access_key_id=Conf.SQS['aws_access_key_id'],
aws_secret_access_key=Conf.SQS['aws_secret_access_key'],
region_name=Conf.SQS['aws_region'])
def get_queue(self):
return self.connection.create_queue(self.list_key)
self.sqs = self.connection.resource('sqs')
return self.sqs.create_queue(QueueName=self.list_key)