support task-priorities when use RabbitMQ / Redis as broker (#189)

* add priority field

* fix ci failed

* add unicode_literals import

* priority default to None

* Happify lint.

* fix ci W504 error

* flake8 ignore W503, W504

Author: b1ngz

django_celery_beat/migrations/0006_periodictask_priority.py

@@ -0,0 +1,24 @@
+# Generated by Django 2.0.6 on 2018-10-22 05:20
+from __future__ import absolute_import, unicode_literals
+
+import django.core.validators
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('django_celery_beat', '0005_add_solarschedule_events_choices_squashed_0009_merge_20181012_1416'), # noqa
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='periodictask',
+            name='priority',
+            field=models.PositiveIntegerField(
+                blank=True,
+                default=None,
+                null=True,
+                validators=[django.core.validators.MaxValueValidator(255)],
+                verbose_name='priority'),
+        ),
+    ]

django_celery_beat/models.py

@@ -7,6 +7,7 @@
 from celery import schedules
 from celery.five import python_2_unicode_compatible
 from django.core.exceptions import MultipleObjectsReturned, ValidationError
+from django.core.validators import MaxValueValidator
 from django.db import models
 from django.db.models import signals
 from django.utils.translation import ugettext_lazy as _
@@ -281,6 +282,10 @@ class PeriodicTask(models.Model):
     routing_key = models.CharField(
         _('routing key'), max_length=200, blank=True, null=True, default=None,
     )
+    priority = models.PositiveIntegerField(
+        _('priority'), default=None, validators=[MaxValueValidator(255)],
+        blank=True, null=True
+    )
     expires = models.DateTimeField(
         _('expires'), blank=True, null=True,
     )

django_celery_beat/schedulers.py

@@ -77,7 +77,8 @@ def __init__(self, model, app=None):
             self._disable(model)
 
         self.options = {}
-        for option in ['queue', 'exchange', 'routing_key', 'expires']:
+        for option in ['queue', 'exchange', 'routing_key', 'expires',
+                       'priority']:
             value = getattr(model, option)
             if value is None:
                 continue
@@ -170,11 +171,12 @@ def _unpack_fields(cls, schedule,
 
     @classmethod
     def _unpack_options(cls, queue=None, exchange=None, routing_key=None,
-                        **kwargs):
+                        priority=None, **kwargs):
         return {
             'queue': queue,
             'exchange': exchange,
             'routing_key': routing_key,
+            'priority': priority
         }
 
     def __repr__(self):
@@ -202,9 +204,9 @@ def __init__(self, *args, **kwargs):
         Scheduler.__init__(self, *args, **kwargs)
         self._finalize = Finalize(self, self.sync, exitpriority=5)
         self.max_interval = (
-            kwargs.get('max_interval') or
-            self.app.conf.beat_max_loop_interval or
-            DEFAULT_MAX_INTERVAL)
+            kwargs.get('max_interval')
+            or self.app.conf.beat_max_loop_interval
+            or DEFAULT_MAX_INTERVAL)
 
     def setup_schedule(self):
         self.install_default_entries(self.schedule)

django_celery_beat/tzcrontab.py

@@ -70,10 +70,10 @@ def __reduce__(self):
 
     def __eq__(self, other):
         if isinstance(other, schedules.crontab):
-            return (other.month_of_year == self.month_of_year and
-                    other.day_of_month == self.day_of_month and
-                    other.day_of_week == self.day_of_week and
-                    other.hour == self.hour and
-                    other.minute == self.minute and
-                    other.tz == self.tz)
+            return (other.month_of_year == self.month_of_year
+                    and other.day_of_month == self.day_of_month
+                    and other.day_of_week == self.day_of_week
+                    and other.hour == self.hour
+                    and other.minute == self.minute
+                    and other.tz == self.tz)
         return NotImplemented

django_celery_beat/utils.py

@@ -39,6 +39,6 @@ def is_database_scheduler(scheduler):
     from kombu.utils import symbol_by_name
     from .schedulers import DatabaseScheduler
     return (
-        scheduler == 'django' or
-        issubclass(symbol_by_name(scheduler), DatabaseScheduler)
+        scheduler == 'django'
+        or issubclass(symbol_by_name(scheduler), DatabaseScheduler)
     )

setup.cfg

@@ -6,7 +6,7 @@ DJANGO_SETTINGS_MODULE=t.proj.settings
 [flake8]
 # classes can be lowercase, arguments and variables can be uppercase
 # whenever it makes the code more readable.
-ignore = N806, N802, N801, N803
+ignore = N806, N802, N801, N803, W503, W504
 
 [pep257]
 ignore = D102,D104,D203,D105,D213

t/unit/test_schedulers.py

@@ -427,10 +427,10 @@ def test_scheduler_schedules_equality_on_change(self, monkeypatch):
 class test_models(SchedulerCase):
 
     def test_IntervalSchedule_unicode(self):
-        assert (text_t(IntervalSchedule(every=1, period='seconds')) ==
-                'every second')
-        assert (text_t(IntervalSchedule(every=10, period='seconds')) ==
-                'every 10 seconds')
+        assert (text_t(IntervalSchedule(every=1, period='seconds'))
+                == 'every second')
+        assert (text_t(IntervalSchedule(every=10, period='seconds'))
+                == 'every 10 seconds')
 
     def test_CrontabSchedule_unicode(self):
         assert text_t(CrontabSchedule(