Adds support for message headers for periodic tasks (#98)

* Add headers field support to execution options

* Add migration for headers on periodictask model

* Fix json loading from model field

* Dumps headers

* Fix migrations for new model field

* Fix for 'priority' field not showing up in django admin execution options

* Fix flake8 errors and future imports on migration file

* Add tests for saving and checking priority field and headers field in options

* Update tests

* Flake 8 fixes

* Fix migrations & dependency tree for PR

* Add python2 compatibility imports and flake8 fixes into migration file

Author: Viatrak

django_celery_beat/admin.py

@@ -137,7 +137,8 @@ class PeriodicTaskAdmin(admin.ModelAdmin):
             'classes': ('extrapretty', 'wide', 'collapse', 'in'),
         }),
         ('Execution Options', {
-            'fields': ('expires', 'queue', 'exchange', 'routing_key'),
+            'fields': ('expires', 'queue', 'exchange', 'routing_key',
+                       'priority', 'headers'),
             'classes': ('extrapretty', 'wide', 'collapse', 'in'),
         }),
     )

django_celery_beat/migrations/0009_periodictask_headers.py

@@ -0,0 +1,25 @@
+# Generated by Django 2.1.5 on 2019-02-09 19:33
+from __future__ import absolute_import, unicode_literals
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('django_celery_beat', '0006_periodictask_priority'),
+        ('django_celery_beat', '0008_auto_20180914_1922'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='periodictask',
+            name='headers',
+            field=models.TextField(
+                blank=True,
+                default='{}',
+                help_text='JSON encoded message headers',
+                verbose_name='Message headers'
+            ),
+        ),
+    ]

django_celery_beat/models.py

@@ -293,6 +293,10 @@ class PeriodicTask(models.Model):
     routing_key = models.CharField(
         _('routing key'), max_length=200, blank=True, null=True, default=None,
     )
+    headers = models.TextField(
+        _('Message headers'), blank=True, default='{}',
+        help_text=_('JSON encoded message headers'),
+    )
     priority = models.PositiveIntegerField(
         _('priority'), default=None, validators=[MaxValueValidator(255)],
         blank=True, null=True
@@ -353,6 +357,7 @@ def save(self, *args, **kwargs):
         self.exchange = self.exchange or None
         self.routing_key = self.routing_key or None
         self.queue = self.queue or None
+        self.headers = self.headers or None
         if not self.enabled:
             self.last_run_at = None
         super(PeriodicTask, self).save(*args, **kwargs)

django_celery_beat/schedulers.py

@@ -86,6 +86,7 @@ def __init__(self, model, app=None):
             if value is None:
                 continue
             self.options[option] = value
+        self.options['headers'] = loads(model.headers or '{}')
 
         self.total_run_count = model.total_run_count
         self.model = model
@@ -197,12 +198,13 @@ def _unpack_fields(cls, schedule,
 
     @classmethod
     def _unpack_options(cls, queue=None, exchange=None, routing_key=None,
-                        priority=None, **kwargs):
+                        priority=None, headers=None, **kwargs):
         return {
             'queue': queue,
             'exchange': exchange,
             'routing_key': routing_key,
-            'priority': priority
+            'priority': priority,
+            'headers': dumps(headers or {}),
         }
 
     def __repr__(self):

t/unit/test_schedulers.py

@@ -97,6 +97,8 @@ def create_model(self, Model=PeriodicTask, **kwargs):
             kwargs='{"callback": "foo"}',
             queue='xaz',
             routing_key='cpu',
+            priority=1,
+            headers='{"_schema_name": "foobar"}',
             exchange='foo',
         )
         return Model(**dict(entry, **kwargs))
@@ -118,6 +120,8 @@ def test_entry(self):
         assert e.options['queue'] == 'xaz'
         assert e.options['exchange'] == 'foo'
         assert e.options['routing_key'] == 'cpu'
+        assert e.options['priority'] == 1
+        assert e.options['headers'] == {'_schema_name': 'foobar'}
 
         right_now = self.app.now()
         m2 = self.create_model_interval(