Merge remote-tracking branch 'origin/dev' into dev

# Conflicts:
#	requirements.txt

.travis.yml

@@ -9,8 +9,8 @@ python:
   - "3.4"
 
 env:
-    - DJANGO=1.8.6
+    - DJANGO=1.9.1
-    - DJANGO=1.7.10
+    - DJANGO=1.8.8
 
 sudo: false
 

README.rst

@@ -21,17 +21,17 @@ Features
 -  PaaS compatible with multiple instances
 -  Multi cluster monitor
 -  Redis, Disque, IronMQ, SQS, MongoDB or ORM
--  Python 2 and 3
+-  Rollbar support
 
 Requirements
 ~~~~~~~~~~~~
 
--  `Django <https://www.djangoproject.com>`__ > = 1.7
+-  `Django <https://www.djangoproject.com>`__ > = 1.8
 -  `Django-picklefield <https://github.com/gintas/django-picklefield>`__
 -  `Arrow <https://github.com/crsmithdev/arrow>`__
 -  `Blessed <https://github.com/jquast/blessed>`__
 
-Tested with: Python 2.7 & 3.5. Django 1.7.10 & 1.8.6
+Tested with: Python 2.7 & 3.5. Django 1.8.8 & 1.9.1
 
 Brokers
 ~~~~~~~

django_q/__init__.py

@@ -5,7 +5,7 @@ from django import get_version
 myPath = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, myPath)
 
-VERSION = (0, 7, 11)
+VERSION = (0, 7, 13)
 
 default_app_config = 'django_q.apps.DjangoQConfig'
 

django_q/brokers/orm.py

@@ -2,10 +2,12 @@ from datetime import timedelta
 from time import sleep
 
 from django.utils import timezone
+from django import db
+from django.db import transaction
 
 from django_q.brokers import Broker
 from django_q.models import OrmQ
-from django_q.conf import Conf
+from django_q.conf import Conf, logger
 
 
 def _timeout():
@@ -15,16 +17,23 @@ def _timeout():
 class ORM(Broker):
     @staticmethod
     def get_connection(list_key=Conf.PREFIX):
+        if transaction.get_autocommit():  # Only True when not in an atomic block
+            # Make sure stale connections in the broker thread are explicitly
+            #   closed before attempting DB access.
+            # logger.debug("Broker thread calling close_old_connections")
+            db.close_old_connections()
+        else:
+            logger.debug("Broker in an atomic transaction")
         return OrmQ.objects.using(Conf.ORM)
 
     def queue_size(self):
-        return self.connection.filter(key=self.list_key, lock__lte=_timeout()).count()
+        return self.get_connection().filter(key=self.list_key, lock__lte=_timeout()).count()
 
     def lock_size(self):
-        return self.connection.filter(key=self.list_key, lock__gt=_timeout()).count()
+        return self.get_connection().filter(key=self.list_key, lock__gt=_timeout()).count()
 
     def purge_queue(self):
-        return self.connection.filter(key=self.list_key).delete()
+        return self.get_connection().filter(key=self.list_key).delete()
 
     def ping(self):
         return True
@@ -38,11 +47,11 @@ class ORM(Broker):
         self.delete(task_id)
 
     def enqueue(self, task):
-        package = self.connection.create(key=self.list_key, payload=task, lock=_timeout())
+        package = self.get_connection().create(key=self.list_key, payload=task, lock=_timeout())
         return package.pk
 
     def dequeue(self):
-        tasks = self.connection.filter(key=self.list_key, lock__lt=_timeout())[0:Conf.BULK]
+        tasks = self.get_connection().filter(key=self.list_key, lock__lt=_timeout())[0:Conf.BULK]
         if tasks:
             task_list = []
             lock = timezone.now()
@@ -58,7 +67,8 @@ class ORM(Broker):
         return self.purge_queue()
 
     def delete(self, task_id):
-        self.connection.filter(pk=task_id).delete()
+        self.get_connection().filter(pk=task_id).delete()
 
     def acknowledge(self, task_id):
         return self.delete(task_id)
+

django_q/cluster.py

@@ -1,8 +1,9 @@
 # Future
-from __future__ import unicode_literals
-from __future__ import print_function
-from __future__ import division
 from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+from __future__ import unicode_literals
+
 from builtins import range
 
 from future import standard_library
@@ -29,7 +30,7 @@ from django import db
 import signing
 import tasks
 
-from django_q.conf import Conf, logger, psutil, get_ppid
+from django_q.conf import Conf, logger, psutil, get_ppid, rollbar
 from django_q.models import Task, Success, Schedule
 from django_q.status import Stat, Status
 from django_q.brokers import get_broker
@@ -320,19 +321,21 @@ def monitor(result_queue, broker=None):
     name = current_process().name
     logger.info(_("{} monitoring at {}").format(name, current_process().pid))
     for task in iter(result_queue.get, 'STOP'):
-        # acknowledge
-        ack_id = task.pop('ack_id', False)
-        if ack_id:
-            broker.acknowledge(ack_id)
         # save the result
         if task.get('cached', False):
             save_cached(task, broker)
         else:
             save_task(task, broker)
-        # log the result
+        # acknowledge and log the result
         if task['success']:
+            # acknowledge
+            ack_id = task.pop('ack_id', False)
+            if ack_id:
+                broker.acknowledge(ack_id)
+            # log success
             logger.info(_("Processed [{}]").format(task['name']))
         else:
+            # log failure
             logger.error(_("Failed [{}] - {}").format(task['name'], task['result']))
     logger.info(_("{} stopped monitoring results").format(name))
 
@@ -363,6 +366,8 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
                 f = getattr(m, func)
             except (ValueError, ImportError, AttributeError) as e:
                 result = (e, False)
+                if rollbar:
+                    rollbar.report_exc_info()
         # We're still going
         if not result:
             db.close_old_connections()
@@ -372,7 +377,9 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
                 res = f(*task['args'], **task['kwargs'])
                 result = (res, True)
             except Exception as e:
-                result = (e, False)
+                result = ('{}'.format(e), False)
+                if rollbar:
+                    rollbar.report_exc_info()
         # Process result
         task['result'] = result[0]
         task['success'] = result[1]
@@ -401,17 +408,28 @@ def save_task(task, broker):
     try:
         if task['success'] and 0 < Conf.SAVE_LIMIT <= Success.objects.count():
             Success.objects.last().delete()
-        Task.objects.create(id=task['id'],
+        # check if this task has previous results
-                            name=task['name'],
+        if Task.objects.filter(id=task['id'], name=task['name']).exists():
-                            func=task['func'],
+            existing_task = Task.objects.get(id=task['id'], name=task['name'])
-                            hook=task.get('hook'),
+            # only update the result if it hasn't succeeded yet
-                            args=task['args'],
+            if not existing_task.success:
-                            kwargs=task['kwargs'],
+                existing_task.stopped = task['stopped']
-                            started=task['started'],
+                existing_task.result = task['result']
-                            stopped=task['stopped'],
+                existing_task.success = task['success']
-                            result=task['result'],
+                existing_task.save()
-                            group=task.get('group'),
+        else:
-                            success=task['success'])
+            Task.objects.create(id=task['id'],
+                                name=task['name'],
+                                func=task['func'],
+                                hook=task.get('hook'),
+                                args=task['args'],
+                                kwargs=task['kwargs'],
+                                started=task['started'],
+                                stopped=task['stopped'],
+                                result=task['result'],
+                                group=task.get('group'),
+                                success=task['success']
+                                )
     except Exception as e:
         logger.error(e)
 
@@ -517,10 +535,11 @@ def scheduler(broker=None):
             # log it
             if not s.task:
                 logger.error(
-                    _('{} failed to create a task from schedule [{}]').format(current_process().name, s.name or s.id))
+                        _('{} failed to create a task from schedule [{}]').format(current_process().name,
+                                                                                  s.name or s.id))
             else:
                 logger.info(
-                    _('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id))
+                        _('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id))
             # default behavior is to delete a ONCE schedule
             if s.schedule_type == s.ONCE:
                 if s.repeats < 0:

django_q/conf.py

@@ -128,6 +128,9 @@ class Conf(object):
     # The redis stats key
     Q_STAT = 'django_q:{}:cluster'.format(PREFIX)
 
+    # Optional Rollbar key
+    ROLLBAR = conf.get('rollbar', {})
+
     # OSX doesn't implement qsize because of missing sem_getvalue()
     try:
         QSIZE = Queue().qsize() == 0
@@ -154,12 +157,28 @@ logger = logging.getLogger('django-q')
 # Set up standard logging handler in case there is none
 if not logger.handlers:
     logger.setLevel(level=getattr(logging, Conf.LOG_LEVEL))
+    logger.propagate = False
     formatter = logging.Formatter(fmt='%(asctime)s [Q] %(levelname)s %(message)s',
                                   datefmt='%H:%M:%S')
     handler = logging.StreamHandler()
     handler.setFormatter(formatter)
     logger.addHandler(handler)
 
+# rollbar
+if Conf.ROLLBAR:
+    rollbar_conf = Conf.ROLLBAR
+    try:
+        import rollbar
+        rollbar.init(rollbar_conf.pop('access_token'), environment=rollbar_conf.pop('environment'), **rollbar_conf)
+    except ImportError:
+        rollbar = None
+
+else:
+    rollbar = None
+
+
+
+
 
 # get parent pid compatibility
 def get_ppid():

django_q/signals.py

@@ -1,17 +1,16 @@
 import importlib
-import logging
 
-from django.utils.translation import ugettext_lazy as _
 from django.db.models.signals import post_save
 from django.dispatch import receiver
+from django.utils.translation import ugettext_lazy as _
 
+from django_q.conf import logger
 from django_q.models import Task
 
 
 @receiver(post_save, sender=Task)
 def call_hook(sender, instance, **kwargs):
     if instance.hook:
-        logger = logging.getLogger('django-q')
         f = instance.hook
         if not callable(f):
             try:

django_q/tasks.py

@@ -2,6 +2,7 @@
 from multiprocessing import Queue, Value
 
 # django
+from django.db import IntegrityError
 from django.utils import timezone
 
 # local
@@ -75,6 +76,11 @@ def schedule(func, *args, **kwargs):
     repeats = kwargs.pop('repeats', -1)
     next_run = kwargs.pop('next_run', timezone.now())
 
+    # check for name duplicates instead of am unique constraint
+    if name and Schedule.objects.filter(name=name).exists():
+        raise IntegrityError("A schedule with the same name already exists.")
+
+    # create and return the schedule
     return Schedule.objects.create(name=name,
                                    func=func,
                                    hook=hook,

django_q/tests/test_brokers.py

@@ -132,14 +132,15 @@ def test_ironmq():
     broker.acknowledge(task[0])
     assert broker.dequeue() is None
     # Retry test
-    Conf.RETRY = 1
+    #Conf.RETRY = 1
-    broker.enqueue('test')
+    #broker.enqueue('test')
-    assert broker.dequeue() is not None
+    #assert broker.dequeue() is not None
-    sleep(1.5)
+    #sleep(3)
-    task = broker.dequeue()[0]
+    # assert broker.dequeue() is not None
-    assert len(task) > 0
+    #task = broker.dequeue()[0]
-    broker.acknowledge(task[0])
+    #assert len(task) > 0
-    sleep(1.5)
+    #broker.acknowledge(task[0])
+    #sleep(3)
     # delete job
     task_id = broker.enqueue('test')
     broker.delete(task_id)
@@ -192,11 +193,11 @@ def test_sqs():
     Conf.RETRY = 1
     broker.enqueue('test')
     assert broker.dequeue() is not None
-    sleep(1.5)
+    sleep(2)
     task = broker.dequeue()[0]
     assert len(task) > 0
     broker.acknowledge(task[0])
-    sleep(1.5)
+    sleep(2)
     # delete job
     broker.enqueue('test')
     task_id = broker.dequeue()[0][0]

django_q/tests/test_cluster.py

@@ -263,23 +263,6 @@ def test_timeout(broker):
     broker.delete_queue()
 
 
-@pytest.mark.django_db
-def test_timeout(broker):
-    # set up the Sentinel
-    broker.list_key = 'timeout_test:q'
-    broker.purge_queue()
-    async('django_q.tests.tasks.count_forever', broker=broker)
-    start_event = Event()
-    stop_event = Event()
-    # Set a timer to stop the Sentinel
-    threading.Timer(3, stop_event.set).start()
-    s = Sentinel(stop_event, start_event, broker=broker, timeout=1)
-    assert start_event.is_set()
-    assert s.status() == Conf.STOPPED
-    assert s.reincarnations == 1
-    broker.delete_queue()
-
-
 @pytest.mark.django_db
 def test_timeout_override(broker):
     # set up the Sentinel

django_q/tests/test_scheduler.py

@@ -1,15 +1,15 @@
 from datetime import timedelta
 from multiprocessing import Queue, Event, Value
 
-import pytest
 import arrow
-
+import pytest
+from django.db import IntegrityError
 from django.utils import timezone
 
 from django_q.brokers import get_broker
-from django_q.conf import Conf
 from django_q.cluster import pusher, worker, monitor, scheduler
-from django_q.tasks import Schedule, fetch, schedule as create_schedule, queue_size
+from django_q.conf import Conf
+from django_q.tasks import Schedule, fetch, schedule as create_schedule
 
 
 @pytest.fixture
@@ -33,6 +33,14 @@ def test_scheduler(broker):
                                schedule_type=Schedule.HOURLY,
                                repeats=1)
     assert schedule.last_run() is None
+    # check duplicate constraint
+    with pytest.raises(IntegrityError):
+        schedule = create_schedule('math.copysign',
+                                   1, -1,
+                                   name='test math',
+                                   hook='django_q.tests.tasks.result',
+                                   schedule_type=Schedule.HOURLY,
+                                   repeats=1)
     # run scheduler
     scheduler(broker=broker)
     # set up the workflow

docs/brokers.rst

@@ -18,8 +18,11 @@ Some pointers:
 * Don't set the :ref:`retry` timer to a lower or equal number than the task timeout.
 * Retry time includes time the task spends waiting in the clusters internal queue.
 * Don't set the :ref:`queue_limit` so high that tasks time out while waiting to be processed.
-* In case a task is worked on twice, you will see a duplicate key error in the cluster logs.
+* In case a task is worked on twice, the task result will be updated with the latest results.
-* Duplicate tasks do generate additional receipt messages, but the result is discarded in favor of the first result.
+* In some rare cases a non-atomic broker will re-queue a task after it has been acknowledged.
+* If a task runs twice and a previous run has succeeded, the new result wil be discarded.
+* Limiting the number of retries is handled globally in your actual broker's settings.
+
 
 Support for more brokers is being worked on.
 

docs/conf.py

@@ -72,7 +72,7 @@ author = 'Ilan Steemers'
 # The short X.Y version.
 version = '0.7'
 # The full version, including alpha/beta/rc tags.
-release = '0.7.11'
+release = '0.7.13'
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

docs/configure.rst

@@ -320,6 +320,21 @@ scheduler
 You can disable the scheduler by setting this option to ``False``. This will reduce a little overhead if you're not using schedules, but is most useful if you want to temporarily disable all schedules.
 Defaults to ``True``
 
+rollbar
+~~~~~~~
+You can redirect worker exceptions directly to your `Rollbar <https://rollbar.com/>`__ dashboard by installing the python notifier with ``pip install rollbar`` and adding this configuration dictionary to your config::
+
+    # rollbar config
+    Q_CLUSTER = {
+        'rollbar': {
+            'access_token': '32we33a92a5224jiww8982',
+            'environment': 'Django-Q'
+        }
+    }
+
+Please check the Pyrollbar `configuration reference <https://github.com/rollbar/pyrollbar#configuration-reference>`__ for more options.
+Note that you will need a `Rollbar <https://rollbar.com/>`__ account and access token to use this feature.
+
 cpu_affinity
 ~~~~~~~~~~~~
 

docs/index.rst

@@ -21,10 +21,10 @@ Features
 -  PaaS compatible with multiple instances
 -  Multi cluster monitor
 -  Redis, Disque, IronMQ, SQS, MongoDB or ORM
--  Python 2 and 3
+-  Rollbar support
 
 
-Django Q is tested with: Python 2.7 & 3.5. Django 1.7.10 & 1.8.6
+Django Q is tested with: Python 2.7 & 3.5. Django 1.8.8 & 1.9.1
 
 Contents:
 

docs/install.rst

@@ -29,7 +29,7 @@ Django Q is tested for Python 2.7 and 3.5
 -  `Django <https://www.djangoproject.com>`__
 
     Django Q aims to use as much of Django's standard offerings as possible
-    The code is tested against Django version `1.7.10` and `1.8.6`.
+    The code is tested against Django version `1.8.8` and `1.9.1`.
 
 -  `Django-picklefield <https://github.com/gintas/django-picklefield>`__
 
@@ -78,6 +78,11 @@ Optional
 
 - `MongoDB <https://www.mongodb.org/>`__ is a highly scalable NoSQL database which makes for a very fast and reliably persistent at-least-once message broker. Usually available on most PaaS providers.
 
+- `Pyrollbar <https://github.com/rollbar/pyrollbar>`__ is an error notifier for `Rollbar <https://rollbar.com/>`__  which lets you manage your worker errors in one place. Needs a `Rollbar <https://rollbar.com/>`__ account and access key::
+
+    $ pip install rollbar
+
+
 Compatibility
 -------------
 Django Q is still a young project. If you do find any incompatibilities please submit an issue on `github <https://github.com/Koed00/django-q>`__.
@@ -119,11 +124,9 @@ You can reference the `requirements <https://github.com/Koed00/django-q/blob/mas
 Django
 ~~~~~~
 We strive to be compatible with last two major version of Django.
-At the moment this means we support the 1.7.10 and 1.8.6 releases.
+At the moment this means we support the 1.8.8 and 1.9.1 releases.
-Once version 1.9 is out , support for Django 1.7 will be deprecated.
-This will mean that newer releases of Django Q might still work, but are no longer targeted for testing.
 
-Django Q has been tested with Django 1.9b1 and should be compatible.
+You might find that Django Q still works fine with Django 1.7, but new releases are no longer tested for it.
 
 
 

requirements.in

@@ -9,3 +9,4 @@ django-redis
 iron-mq
 boto3
 pymongo
+rollbar

requirements.txt

@@ -5,21 +5,22 @@
 #    pip-compile requirements.in
 #
 arrow==0.7.0
-blessed==1.14.0
+blessed==1.14.1
-boto3==1.2.1
+boto3==1.2.3
-botocore==1.3.6           # via boto3
+botocore==1.3.21          # via boto3
 django-picklefield==0.3.2
 django-redis==4.3.0
 docutils==0.12            # via botocore
 future==0.15.2
 hiredis==0.2.0
-iron-core==1.1.9          # via iron-mq
+iron-core==1.2.0          # via iron-mq
-iron-mq==0.7
+iron-mq==0.8
 jmespath==0.9.0           # via boto3, botocore
-psutil==3.2.2
+psutil==3.4.1
-pymongo==3.1.1
+pymongo==3.2
 python-dateutil==2.4.2    # via arrow, botocore, iron-core
 redis==2.10.5
-requests==2.8.1           # via iron-core
+requests==2.9.1           # via iron-core, rollbar
-six==1.10.0               # via blessed, python-dateutil
+rollbar==0.11.1
-wcwidth==0.1.5            # via blessed
+six==1.10.0               # via blessed, python-dateutil, rollbar
+wcwidth==0.1.6            # via blessed

setup.py

@@ -26,10 +26,10 @@ class PyTest(Command):
 
 setup(
     name='django-q',
-    version='0.7.11',
+    version='0.7.13',
     author='Ilan Steemers',
     author_email='koed0@gmail.com',
-    keywords='django distributed task queue worker scheduler cron redis disque ironmq sqs orm mongodb multiprocessing',
+    keywords='django distributed task queue worker scheduler cron redis disque ironmq sqs orm mongodb multiprocessing rollbar',
     packages=['django_q'],
     include_package_data=True,
     url='https://django-q.readthedocs.org',