diff --git a/README.rst b/README.rst index 1a21b8b..9e0f254 100644 --- a/README.rst +++ b/README.rst @@ -20,7 +20,7 @@ Features - Django Admin integration - PaaS compatible with multiple instances - Multi cluster monitor -- Redis, Disque, IronMQ or SQS +- Redis, Disque, IronMQ, SQS or ORM - Python 2 and 3 Requirements @@ -35,11 +35,11 @@ Tested with: Python 2.7 & 3.4. Django 1.7.10 & 1.8.4 Brokers ~~~~~~~ -- `Redis `__ -- `Disque `__ -- `IronMQ `__ -- `Amazon SQS `__ - +- `Redis `__ +- `Disque `__ +- `IronMQ `__ +- `Amazon SQS `__ +- `Django ORM `__ Installation ~~~~~~~~~~~~ diff --git a/django_q/admin.py b/django_q/admin.py index f03e90f..57f22d5 100644 --- a/django_q/admin.py +++ b/django_q/admin.py @@ -3,11 +3,11 @@ from django.contrib import admin from django.utils.translation import ugettext_lazy as _ from .tasks import async -from .models import Success, Failure, Schedule +from .models import Success, Failure, Schedule, OrmQ +from .conf import Conf class TaskAdmin(admin.ModelAdmin): - """model admin for success tasks.""" list_display = ( @@ -34,8 +34,8 @@ class TaskAdmin(admin.ModelAdmin): def get_readonly_fields(self, request, obj=None): """Set all fields readonly.""" - return list(self.readonly_fields) +\ - [field.name for field in obj._meta.fields] + return list(self.readonly_fields) + \ + [field.name for field in obj._meta.fields] def retry_failed(FailAdmin, request, queryset): @@ -49,7 +49,6 @@ retry_failed.short_description = _("Resubmit selected tasks to queue") class FailAdmin(admin.ModelAdmin): - """model admin for failed tasks.""" list_display = ( @@ -72,11 +71,10 @@ class FailAdmin(admin.ModelAdmin): def get_readonly_fields(self, request, obj=None): """Set all fields readonly.""" return list(self.readonly_fields) + \ - [field.name for field in obj._meta.fields] + [field.name for field in obj._meta.fields] class ScheduleAdmin(admin.ModelAdmin): - """ model admin for schedules """ list_display = ( @@ -95,6 +93,21 @@ class ScheduleAdmin(admin.ModelAdmin): list_display_links = ('id', 'name') +class QueueAdmin(admin.ModelAdmin): + """ queue admin for ORM broker """ + list_display = ( + 'id', + 'key', + 'lock' + ) + + def has_add_permission(self, request, obj=None): + """Don't allow adds.""" + return False + admin.site.register(Schedule, ScheduleAdmin) admin.site.register(Success, TaskAdmin) admin.site.register(Failure, FailAdmin) + +if Conf.ORM or Conf.TESTING: + admin.site.register(OrmQ, QueueAdmin) diff --git a/django_q/brokers/__init__.py b/django_q/brokers/__init__.py index fed4590..0b5b84f 100644 --- a/django_q/brokers/__init__.py +++ b/django_q/brokers/__init__.py @@ -160,6 +160,9 @@ def get_broker(list_key=Conf.PREFIX): elif Conf.SQS: from brokers import aws_sqs return aws_sqs.Sqs(list_key=list_key) + elif Conf.ORM: + from brokers import orm + return orm.ORM(list_key=list_key) # default to redis else: from brokers import redis_broker diff --git a/django_q/brokers/orm.py b/django_q/brokers/orm.py new file mode 100644 index 0000000..461f894 --- /dev/null +++ b/django_q/brokers/orm.py @@ -0,0 +1,62 @@ +from datetime import timedelta +from time import sleep +from django.utils import timezone +from django.db.models import Q +from django_q.brokers import Broker +from django_q.models import OrmQ +from django_q.conf import Conf + + +class ORM(Broker): + def queue_size(self): + return OrmQ.objects.using(Conf.ORM) \ + .filter(Q(key=self.list_key, lock__isnull=True) | + Q(key=self.list_key, lock__lte=timezone.now() - timedelta(seconds=Conf.RETRY))) \ + .count() + + def purge_queue(self): + return OrmQ.objects.using(Conf.ORM).filter(key=self.list_key).delete() + + def ping(self): + return True + + def info(self): + return 'ORM {}'.format(Conf.ORM) + + def fail(self, task_id): + self.delete(task_id) + + def enqueue(self, task): + package = OrmQ.objects.using(Conf.ORM).create(key=self.list_key, payload=task) + return package.pk + + def dequeue(self): + if len(self.task_cache) > 0: + t = self.task_cache.pop() + return t.pk, t.payload + else: + # Get new and timed out tasks + tasks = OrmQ.objects.using(Conf.ORM).filter( + Q(key=self.list_key, lock__isnull=True) | + Q(key=self.list_key, lock__lte=timezone.now() - timedelta(seconds=Conf.RETRY)))[:Conf.BULK] + if tasks: + # lock them + OrmQ.objects.using(Conf.ORM).filter(pk__in=tasks).update(lock=timezone.now()) + tasks = [t for t in tasks] + # pop one task + t = tasks.pop() + if tasks: + # add remainder to cache + self.task_cache = [t for t in tasks] + return t.pk, t.payload + # empty queue, spare the cpu + sleep(0.2) + + def delete_queue(self): + return self.purge_queue() + + def delete(self, task_id): + return OrmQ.objects.using(Conf.ORM).filter(pk=task_id).delete() + + def acknowledge(self, task_id): + return self.delete(task_id) diff --git a/django_q/cluster.py b/django_q/cluster.py index 13142a6..8e99fab 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -238,7 +238,7 @@ class Sentinel(object): self.reincarnate(self.pusher) # Call scheduler once a minute (or so) counter += cycle - if counter == 30: + if counter == 30 and Conf.SCHEDULER: counter = 0 scheduler(broker=self.broker) # Save current status diff --git a/django_q/conf.py b/django_q/conf.py index 3984a28..b312c4f 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -47,6 +47,9 @@ class Conf(object): # SQS broker SQS = conf.get('sqs', None) + # ORM broker + ORM = conf.get('orm', None) + # Name of the cluster or site. For when you run multiple sites on one redis server PREFIX = conf.get('name', 'default') @@ -57,6 +60,9 @@ class Conf(object): # Failures are always saved SAVE_LIMIT = conf.get('save_limit', 250) + # Disable the scheduler + SCHEDULER = conf.get('scheduler', True) + # Number of workers in the pool. Default is cpu count if implemented, otherwise 4. WORKERS = conf.get('workers', False) if not WORKERS: diff --git a/django_q/migrations/0007_ormq.py b/django_q/migrations/0007_ormq.py new file mode 100644 index 0000000..1c42dc2 --- /dev/null +++ b/django_q/migrations/0007_ormq.py @@ -0,0 +1,27 @@ +# -*- coding: utf-8 -*- +from __future__ import unicode_literals + +from django.db import models, migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ('django_q', '0006_auto_20150805_1817'), + ] + + operations = [ + migrations.CreateModel( + name='OrmQ', + fields=[ + ('id', models.AutoField(primary_key=True, auto_created=True, verbose_name='ID', serialize=False)), + ('key', models.CharField(max_length=100)), + ('payload', models.TextField()), + ('lock', models.DateTimeField(null=True)), + ], + options={ + 'verbose_name_plural': 'Queued tasks', + 'verbose_name': 'Queued task', + }, + ), + ] diff --git a/django_q/models.py b/django_q/models.py index d42d4af..de05408 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -188,6 +188,17 @@ class Schedule(models.Model): ordering = ['next_run'] +class OrmQ(models.Model): + key = models.CharField(max_length=100) + payload = models.TextField() + lock = models.DateTimeField(null=True) + + class Meta: + app_label = 'django_q' + verbose_name = _('Queued task') + verbose_name_plural = _('Queued tasks') + + # Backwards compatibility for Django 1.7 def decode_results(values): if get_version().split('.')[1] == '7': diff --git a/django_q/tests/test_admin.py b/django_q/tests/test_admin.py index a18bcdf..b82ad2f 100644 --- a/django_q/tests/test_admin.py +++ b/django_q/tests/test_admin.py @@ -4,12 +4,14 @@ from django.utils import timezone import pytest from django_q.tasks import schedule -from django_q.models import Task, Failure +from django_q.models import Task, Failure, OrmQ from django_q.humanhash import uuid +from django_q.conf import Conf @pytest.mark.django_db def test_admin_views(admin_client): + Conf.ORM='default' s = schedule('sched.test') tag = uuid() f = Task.objects.create( @@ -27,6 +29,9 @@ def test_admin_views(admin_client): started=timezone.now(), stopped=timezone.now(), success=True) + q = OrmQ.objects.create( + key='test', + payload='test') admin_urls = ( # schedule reverse('admin:django_q_schedule_changelist'), @@ -44,6 +49,11 @@ def test_admin_views(admin_client): reverse('admin:django_q_failure_change', args=(f.id,)), reverse('admin:django_q_failure_history', args=(f.id,)), reverse('admin:django_q_failure_delete', args=(f.id,)), + # orm queue + reverse('admin:django_q_ormq_changelist'), + reverse('admin:django_q_ormq_change', args=(q.id,)), + reverse('admin:django_q_ormq_history', args=(q.id,)), + reverse('admin:django_q_ormq_delete', args=(q.id,)), ) for url in admin_urls: @@ -57,3 +67,6 @@ def test_admin_views(admin_client): response = admin_client.post(url, data) assert response.status_code == 302 assert Failure.objects.filter(name=f.id).exists() is False + # cleanup + q.delete() + Conf.ORM = None diff --git a/django_q/tests/test_brokers.py b/django_q/tests/test_brokers.py index 96c07f8..350c2a0 100644 --- a/django_q/tests/test_brokers.py +++ b/django_q/tests/test_brokers.py @@ -215,3 +215,58 @@ def test_sqs(): Conf.SQS = None Conf.BULK = 1 Conf.DJANGO_REDIS = 'default' + +@pytest.mark.django_db +def test_orm(): + Conf.ORM = 'default' + # check broker + broker = get_broker(list_key='orm_test') + assert broker.ping() is True + assert broker.info() is not None + # clear before we start + broker.delete_queue() + # enqueue + broker.enqueue('test') + assert broker.queue_size() == 1 + # dequeue + task = broker.dequeue() + assert task[1] == 'test' + broker.acknowledge(task[0]) + assert broker.queue_size() == 0 + # Retry test + Conf.RETRY = 1 + broker.enqueue('test') + assert broker.queue_size() == 1 + broker.dequeue() + assert broker.queue_size() == 0 + sleep(1.5) + assert broker.queue_size() == 1 + task = broker.dequeue() + assert broker.queue_size() == 0 + broker.acknowledge(task[0]) + sleep(1.5) + assert broker.queue_size() == 0 + # delete job + task_id = broker.enqueue('test') + broker.delete(task_id) + assert broker.dequeue() is None + # fail + task_id = broker.enqueue('test') + broker.fail(task_id) + # bulk test + for i in range(5): + broker.enqueue('test') + Conf.BULK = 5 + for i in range(5): + task = broker.dequeue() + assert task is not None + broker.acknowledge(task[0]) + # test duplicate acknowledge + broker.acknowledge(task[0]) + # delete queue + broker.enqueue('test') + broker.enqueue('test') + broker.delete_queue() + assert broker.queue_size() == 0 + # back to django-redis + Conf.ORM = None diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index dcd5394..ceabb72 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -32,6 +32,7 @@ def broker(): Conf.DISQUE_NODES = None Conf.IRON_MQ = None Conf.SQS = None + Conf.ORM = None Conf.DJANGO_REDIS = 'default' return get_broker() diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index 4cf7fd3..a8e5abc 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -17,6 +17,7 @@ def broker(): Conf.DISQUE_NODES = None Conf.IRON_MQ = None Conf.SQS = None + Conf.ORM = None Conf.DJANGO_REDIS = 'default' return get_broker() diff --git a/docs/admin.rst b/docs/admin.rst index fb6da85..9eace11 100644 --- a/docs/admin.rst +++ b/docs/admin.rst @@ -81,3 +81,9 @@ Indicates the success status of the last scheduled task, if any. Uses the :class:`Schedule` model + +Queued tasks +------------ +This admin view is only enabled when you use the :ref:`orm_broker` broker. +It shows all tasks packages currently in the broker queue. The ``lock`` column shows the moment at which this package was picked up by the cluster and is used to determine whether it has expired or not. +For development purposes you can edit and delete queued tasks from here. diff --git a/docs/brokers.rst b/docs/brokers.rst index 396cf0b..a3fe3e4 100644 --- a/docs/brokers.rst +++ b/docs/brokers.rst @@ -73,6 +73,19 @@ Although `SQS `__ is not the fastest, it is stable, * Requires the `boto3 `__ client library: ``pip install boto3`` * See the :ref:`sqs_configuration` configuration section for options. +.. _orm_broker: + +Django ORM +---------- +Select this to use Django's database backend as a message broker. +Unless you have configured a dedicated database backend for it, this should probably not be your first choice for a high traffic setup. +However for a medium message rate and scheduled tasks, this is the most convenient guaranteed delivery broker. + +* Delivery receipts +* Supports bulk dequeue +* Needs Django's `Cache framework `__ configured for monitoring +* Queue editable in Django Admin +* See the :ref:`orm_configuration` configuration on how to set it up. Reference --------- diff --git a/docs/conf.py b/docs/conf.py index d82220c..a6ee0a9 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -70,9 +70,9 @@ author = 'Ilan Steemers' # built documents. # # The short X.Y version. -version = '0.6' +version = '0.7' # The full version, including alpha/beta/rc tags. -release = '0.6.4' +release = '0.7.0' # The language for content autogenerated by Sphinx. Refer to documentation # for a list of supported languages. diff --git a/docs/configure.rst b/docs/configure.rst index 775b470..90432d6 100644 --- a/docs/configure.rst +++ b/docs/configure.rst @@ -241,6 +241,29 @@ Please make sure these credentials have proper SQS access. Amazon SQS only supports a bulk setting between 1 and 10, with the total payload not exceeding 256kb. +.. _orm_configuration: + +orm +~~~ +If you want to use Django's database backend as a message broker, set the ``orm`` keyword to the database connection you want it to use:: + + # example ORM broker connection + + Q_CLUSTER = { + 'name': 'DjangORM', + 'workers': 4, + 'timeout': 90, + 'retry': 120, + 'queue_limit': 50, + 'bulk': 10, + 'orm': 'default' + } + +Using the Django ORM backend will also enable the Queued Tasks table in the Admin. + +If you need better performance , you should consider using a different database backend than the main project. +Set ``orm`` to the name of that database connection and make sure you run migrations on it using the ``--database`` option. + .. _bulk: bulk @@ -257,6 +280,11 @@ cache For some brokers, you will need to set up the Django `cache framework `__ to gather statistics for the monitor. You can indicate which cache to use by setting this value. Defaults to ``default``. +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`` + cpu_affinity ~~~~~~~~~~~~ diff --git a/docs/index.rst b/docs/index.rst index 4fc8886..ced20cc 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -20,7 +20,7 @@ Features - Django Admin integration - PaaS compatible with multiple instances - Multi cluster monitor -- Redis, Disque, IronMQ or SQS +- Redis, Disque, IronMQ, SQS or ORM - Python 2 and 3