adding boto3 sqs retry limit, fixing tests and various python3 issues

Author: Alexey Tsitkin

README.md

@@ -133,7 +133,7 @@ The following settings can be used to fine tune django-eb-sqs. Copy them into yo
 - EB_AWS_REGION (`us-east-1`): The AWS region to use when working with SQS.
 - EB_SQS_MAX_NUMBER_OF_MESSAGES (`10`): The maximum number of messages to read in a single call from SQS (<= 10).
 - EB_SQS_WAIT_TIME_S (`2`): The time to wait (seconds) when receiving messages from SQS.
-- EB_SQS_AUTO_ADD_QUEUE (`True`): If queues should be added automatically to AWS if they don't exist.
+- EB_SQS_AUTO_ADD_QUEUE (`False`): If queues should be added automatically to AWS if they don't exist.
 - EB_SQS_DEAD_LETTER_MODE (`False`): Enable if this worker is handling the SQS dead letter queue. Tasks won't be executed but group callback is.
 - EB_SQS_DEFAULT_DELAY (`0`): Default task delay time in seconds.
 - EB_SQS_DEFAULT_MAX_RETRIES (`0`): Default retry limit for all tasks.
@@ -147,6 +147,7 @@ The following settings can be used to fine tune django-eb-sqs. Copy them into yo
 - EB_SQS_REDIS_EXPIRY (`604800`): Default expiry time in seconds until a group is removed
 - EB_SQS_REDIS_KEY_PREFIX (`eb-sqs-`): Prefix used for all Redis keys
 - EB_SQS_USE_PICKLE (`False`): Enable to use `pickle` to serialize task parameters. Uses `json` as default.
+- EB_SQS_AWS_MAX_RETRIES (`10`): Default retry limit on a boto3 call to AWS SQS.
 
 
 ### Development
@@ -158,5 +159,5 @@ Make sure to install the development dependencies from `development.txt`.
 The build in tests can be executed with the Django test runner.
 
 ```bash
-django-admin test --settings=eb_sqs.test_settings
+python -m django test --settings=eb_sqs.test_settings
 ```

development.txt

@@ -1,4 +1,4 @@
-boto3==1.4.4
+boto3==1.4.7
 Django==1.10.6
 mock==2.0.0
 moto==0.4.24

eb_sqs/management/commands/process_queue.py

@@ -3,6 +3,7 @@
 import boto3
 import logging
 
+from botocore.config import Config
 from django.core.management import BaseCommand, CommandError
 
 from eb_sqs import settings
@@ -28,7 +29,11 @@ def handle(self, *args, **options):
 
         logger.debug('[django-eb-sqs] Connecting to SQS: {}'.format(', '.join(queue_names)))
 
-        sqs = boto3.resource('sqs', region_name=settings.AWS_REGION)
+        sqs = boto3.resource(
+            'sqs',
+            region_name=settings.AWS_REGION,
+            config=Config(retries={'max_attempts': settings.AWS_MAX_RETRIES})
+        )
         queues = [sqs.get_queue_by_name(QueueName=queue_name) for queue_name in queue_names]
 
         logger.debug('[django-eb-sqs] Connected to SQS: {}'.format(', '.join(queue_names)))

eb_sqs/management/commands/run_eb_sqs_worker.py

@@ -1,6 +1,8 @@
 from __future__ import absolute_import, unicode_literals
 
 import boto3
+from botocore.config import Config
+from django.conf import settings
 from django.core.management import BaseCommand, CommandError
 import requests
 from requests.exceptions import ConnectionError
@@ -36,7 +38,11 @@ def handle(self, *args, **options):
 
         try:
             self.stdout.write('Connect to SQS')
-            sqs = boto3.resource("sqs")
+            sqs = boto3.resource(
+                'sqs',
+                region_name=settings.AWS_REGION,
+                config=Config(retries={'max_attempts': settings.AWS_MAX_RETRIES})
+            )
             queue = sqs.get_queue_by_name(QueueName=queue_name)
             self.stdout.write('> Connected')
 

eb_sqs/settings.py

@@ -2,14 +2,14 @@
 
 from django.conf import settings
 
-AWS_REGION = getattr(settings, 'EB_AWS_REGION', 'us-east-1')  # type: unicode
+AWS_REGION = getattr(settings, 'EB_AWS_REGION', 'us-east-1')  # type: str
 
 MAX_NUMBER_OF_MESSAGES = getattr(settings, 'EB_SQS_MAX_NUMBER_OF_MESSAGES', 10)  # type: int
 WAIT_TIME_S = getattr(settings, 'EB_SQS_WAIT_TIME_S', 2)  # type: int
 
-AUTO_ADD_QUEUE = getattr(settings, 'EB_SQS_AUTO_ADD_QUEUE', True)  # type: bool
-QUEUE_PREFIX = getattr(settings, 'EB_SQS_QUEUE_PREFIX', 'eb-sqs-')  # type: unicode
-DEFAULT_QUEUE = getattr(settings, 'EB_SQS_DEFAULT_QUEUE', 'default')  # type: unicode
+AUTO_ADD_QUEUE = getattr(settings, 'EB_SQS_AUTO_ADD_QUEUE', False)  # type: bool
+QUEUE_PREFIX = getattr(settings, 'EB_SQS_QUEUE_PREFIX', 'eb-sqs-')  # type: str
+DEFAULT_QUEUE = getattr(settings, 'EB_SQS_DEFAULT_QUEUE', 'default')  # type: str
 
 EXECUTE_INLINE = getattr(settings, 'EB_SQS_EXECUTE_INLINE', False)  # type: bool
 FORCE_SERIALIZATION = getattr(settings, 'EB_SQS_FORCE_SERIALIZATION', False)  # type: bool
@@ -25,8 +25,10 @@
 REDIS_CLIENT = getattr(settings, 'EB_SQS_REDIS_CLIENT', None)  # type: StrictRedis
 # default: 7 days
 REDIS_EXPIRY = getattr(settings, 'EB_SQS_REDIS_EXPIRY', 3600 * 24 * 7)  # type: int
-REDIS_KEY_PREFIX = getattr(settings, 'EB_SQS_REDIS_KEY_PREFIX', 'eb-sqs-')  # type: string
+REDIS_KEY_PREFIX = getattr(settings, 'EB_SQS_REDIS_KEY_PREFIX', 'eb-sqs-')  # type: str
 
 WORKER_FACTORY = getattr(settings, 'EB_SQS_WORKER_FACTORY', None)  # type: WorkerFactory
 
 DEAD_LETTER_MODE = getattr(settings, 'EB_SQS_DEAD_LETTER_MODE', False)  # type: bool
+
+AWS_MAX_RETRIES = getattr(settings, 'EB_SQS_AWS_MAX_RETRIES', 10)  # type: int

eb_sqs/tests/worker/tests_worker_task.py

@@ -1,5 +1,6 @@
 from __future__ import absolute_import, unicode_literals
 
+import json
 from unittest import TestCase
 
 from eb_sqs.worker.worker_task import WorkerTask
@@ -24,7 +25,7 @@ def test_serialize_worker_task(self):
         worker_task = WorkerTask('id-1', 'group-5', 'default', dummy_function, [], {}, 5, 0, 'retry-uuid', False)
         msg = worker_task.serialize()
 
-        self.assertEqual(msg, self.dummy_msg)
+        self.assertDictEqual(json.loads(msg), json.loads(self.dummy_msg))
 
     def test_deserialize_worker_task(self):
         worker_task = WorkerTask.deserialize(self.dummy_msg)

eb_sqs/worker/worker.py

@@ -4,6 +4,8 @@
 import logging
 import uuid
 
+from six import string_types
+
 from eb_sqs import settings
 from eb_sqs.worker.group_client import GroupClient
 from eb_sqs.worker.queue_client import QueueDoesNotExistException, QueueClient, QueueClientException
@@ -163,7 +165,7 @@ def _execute_group_callback(self, worker_task):
         if settings.GROUP_CALLBACK_TASK:
             callback = settings.GROUP_CALLBACK_TASK
 
-            if isinstance(callback, basestring):
+            if isinstance(callback, string_types):
                 func_name = callback.split(".")[-1]
                 func_path = ".".join(callback.split(".")[:-1])
                 func_module = importlib.import_module(func_path)

eb_sqs/worker/worker_task.py

@@ -75,7 +75,7 @@ def copy(self, use_serialization):
     @staticmethod
     def _pickle_args(args):
         # type: (dict) -> unicode
-        return base64.b64encode(pickle.dumps(args, pickle.HIGHEST_PROTOCOL))
+        return base64.b64encode(pickle.dumps(args, pickle.HIGHEST_PROTOCOL)).decode('utf-8')
 
     @staticmethod
     def deserialize(msg):
@@ -105,4 +105,4 @@ def deserialize(msg):
     @staticmethod
     def _unpickle_args(args):
         # type: (unicode) -> dict
-        return pickle.loads(base64.b64decode(args))
+        return pickle.loads(base64.b64decode(args.encode('utf-8')))

setup.py

@@ -14,7 +14,7 @@
     long_description=README,
     url='https://github.com/cuda-networks/django-eb-sqs',
     install_requires=[
-        'boto3>=1.3.1',
+        'boto3>=1.4.7',
         'Django>=1.7',
         'redis>=2.10',
         'requests>=2',