QueueFetcher makes dealing with SQS queues in Django easier
Project description
QueueFetcher allows you to deal with Amazon SQS queues in an easier manner in Django.
It provides:
run_queue management task to start the task from cli
QueueFetcher class to do the heavy lifting with the pieces seperated out and testable
Getting started
Install queue-fetcher from pip
Add queue_fetcher to INSTALLED_APPLICATIONS
Add to your settings.py:
TEST_SQS = False
QUEUES = {
'Internal Name': 'Name On Amazon'
}
Now build your tasks in your tasks package:
from queue_fetcher.tasks import QueueFetcher
class SampleQueueTask(QueueFetcher):
queue = 'test'
def process_sample(self, msg):
raise NotImplementedError('This does nothing.. yet')
QueueFetcher expects messages from SQS to contain a list of events, with each event containing a message_type attribute of something like update_transaction.
This is then dispatched to a function prefixed with process_.
Visibility Timeout
Tasks run from Django Queue Fetcher can, when they hit an error, keep thrashing your SQS queues. We recommend you set a visibility_timeout on each task to minimise your costs and any errors you send to your logs:
from queue_fetcher.tasks import QueueFetcher
class MyQueueFetcher(QueueFetcher):
"""Same Queue Fetcher.
"""
queue = 'test'
visibility_timeout = 60 # Tell SQS to give us 60 seconds to process
def process_my_message(self, msg):
"""Process a message.
"""
return True
Testing your Code
The queue-fetcher app includes a QueueTestCase class that removes the need to handle SQS in your test code. To use it, simple extend the class and use get_yaml or get_json to get your fixtures, located in the same app as your test.
from queue_fetcher.test import QueueTestCase
from .tasks import ExampleTaskClass
class ExampleTestCase(QueueTestCase):
"""
"""
def test_my_app(self):
"""
"""
fixture = self.get_json('exampleapp/test.sqs.json')
task = ExampleTaskClass()
task.read(fixture)
# Insert your assertions here
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
Built Distributions
Hashes for queue_fetcher-2.0.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 9ac798133a5cd0ea63064d29d9ce38af14b94ca8072081295c4112ec8678dfb2 |
|
MD5 | e3c66145209ac0feff818d6a56b33d66 |
|
BLAKE2b-256 | 17c3f82724f781d1662ce75b8084989284126e094f7dcdd4b8f7a48d4150d7de |
Hashes for queue_fetcher-2.0.0-py2.py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 52f1c9fabe9394c5a267de9a4715d6c1ed783ccd20aa92b00dc901245525bb2d |
|
MD5 | d99563510c943c2823bd04da58a270c1 |
|
BLAKE2b-256 | b444cae6599d9be442cdcb4fce258c8a27d224ef139a4d1178fc08b7bb90432d |