#360 - adding classmethods to DatabaseScheduler for simple dynamic task ... (#361)

* #360 - adding classmethods to DatabaseScheduler for simple dynamic task manipulations

* Update Changelog

Author: ldmberman

Changelog

@@ -41,6 +41,10 @@
 
     Fix contributed by Vytis Banaitis.
 
+- Added friendly API for dynamically creating/updating/deleting tasks
+
+    Fix contributed by Lev Berman
+
 .. _version-3.1.17:
 
 3.1.17

djcelery/schedulers.py

@@ -254,3 +254,17 @@ def schedule(self):
                     repr(entry) for entry in itervalues(self._schedule)),
                 )
         return self._schedule
+
+    @classmethod
+    def create_or_update_task(cls, name, **schedule_dict):
+        if 'schedule' not in schedule_dict:
+            try:
+                schedule_dict['schedule'] = \
+                    PeriodicTask._default_manager.get(name=name).schedule
+            except PeriodicTask.DoesNotExist:
+                pass
+        cls.Entry.from_entry(name, **schedule_dict)
+
+    @classmethod
+    def delete_task(cls, name):
+        PeriodicTask._default_manager.get(name=name).delete()

djcelery/tests/test_schedulers.py

@@ -214,6 +214,35 @@ def test_sync_rollback_on_save_error(self):
         self.s._dirty.add(self.m1.name)
         self.assertRaises(RuntimeError, self.s.sync)
 
+    def test_dynamic_create_update_delete_task(self):
+        schedule_dict = {
+            'task': 'some_task',
+            'schedule': timedelta(seconds=5),
+            'args': ('arg1', 'arg2'),
+        }
+        task1_name = 'test_task 1'
+        task2_name = 'test_task 2'
+        schedulers.DatabaseScheduler.create_or_update_task(task1_name,
+                                                           **schedule_dict)
+        schedulers.DatabaseScheduler.create_or_update_task(task2_name,
+                                                           **schedule_dict)
+        PeriodicTask.objects.get(name=task1_name)  # assert not raises
+        PeriodicTask.objects.get(name=task2_name)
+        schedulers.DatabaseScheduler.create_or_update_task(task1_name,
+                                                           args=('arg3',))
+        self.assertEqual(PeriodicTask.objects.get(name=task1_name).args,
+                         u'["arg3"]')
+        schedulers.DatabaseScheduler.create_or_update_task(
+            task2_name, schedule=timedelta(10))
+        self.assertEqual(PeriodicTask.objects.get(name=task2_name).schedule,
+                         timedelta(10))
+        schedulers.DatabaseScheduler.delete_task(task1_name)
+        schedulers.DatabaseScheduler.delete_task(task2_name)
+        self.assertRaises(PeriodicTask.DoesNotExist,
+                          lambda: PeriodicTask.objects.get(name=task1_name))
+        self.assertRaises(PeriodicTask.DoesNotExist,
+                          lambda: PeriodicTask.objects.get(name=task2_name))
+
 
 class test_models(unittest.TestCase):