added run action to periodic tasks admin (#52)

* added run action to periodic tasks admin

* fixed flake8 line length issues

* Added tests and updated changelog

* map call changed to list comprehension
* added self to AUTHORS

* Update .travis.yml

need rabbitmq service for CI

* fixed flake8 line length issues

Author: bradshjg

.travis.yml

@@ -27,6 +27,7 @@ env:
     - TOXENV=pydocstyle
     - TOXENV=cov
 install: travis_retry pip install -U tox
+services: rabbitmq
 script: tox -v -- -v
 after_success:
   - .tox/$TRAVIS_PYTHON_VERSION/bin/coverage xml

AUTHORS

@@ -46,6 +46,7 @@ Jeffrey Hu <[email protected]>
 Jens Alm <[email protected]>
 Jerzy Kozera <[email protected]>
 Jesper Noehr <[email protected]>
+Jimmy Bradshaw <[email protected]>
 Joey Wilhelm <[email protected]>
 John Andrews <johna@stjit011.(none)>
 John Watson <[email protected]>

Changelog

@@ -4,6 +4,15 @@
  Change history
 ================
 
+.. _version-1.1.0:
+
+1.1.0
+=====
+:release-date: TODO
+:release-by: TODO
+
+- Adds "run now" admin action for tasks.
+
 .. _version-1.0.1:
 
 1.0.1

django_celery_beat/admin.py

@@ -111,12 +111,13 @@ def clean_kwargs(self):
 
 
 class PeriodicTaskAdmin(admin.ModelAdmin):
-    """Admin-interface for peridic tasks."""
+    """Admin-interface for periodic tasks."""
 
     form = PeriodicTaskForm
     model = PeriodicTask
+    celery_app = current_app
     list_display = ('__str__', 'enabled')
-    actions = ('enable_tasks', 'disable_tasks')
+    actions = ('enable_tasks', 'disable_tasks', 'run_tasks')
     fieldsets = (
         (None, {
             'fields': ('name', 'regtask', 'task', 'enabled'),
@@ -173,6 +174,25 @@ def disable_tasks(self, request, queryset):
         )
     disable_tasks.short_description = _('Disable selected tasks')
 
+    def run_tasks(self, request, queryset):
+        self.celery_app.loader.import_default_modules()
+        tasks = [(self.celery_app.tasks.get(task.task),
+                  loads(task.args),
+                  loads(task.kwargs))
+                 for task in queryset]
+        task_ids = [task.delay(*args, **kwargs)
+                    for task, args, kwargs in tasks]
+        tasks_run = len(task_ids)
+        self.message_user(
+            request,
+            _('{0} task{1} {2} successfully run').format(
+                tasks_run,
+                pluralize(tasks_run),
+                pluralize(tasks_run, _('was,were')),
+            ),
+        )
+    run_tasks.short_description = _('Run selected tasks')
+
 
 admin.site.register(IntervalSchedule)
 admin.site.register(CrontabSchedule)

t/unit/test_schedulers.py

@@ -5,10 +5,15 @@
 from datetime import datetime, timedelta
 from itertools import count
 
+from django.contrib.admin.sites import AdminSite
+from django.contrib.messages.storage.fallback import FallbackStorage
+from django.test import RequestFactory
+
 from celery.five import monotonic, text_t
 from celery.schedules import schedule, crontab
 
 from django_celery_beat import schedulers
+from django_celery_beat.admin import PeriodicTaskAdmin
 from django_celery_beat.models import (
     PeriodicTask, PeriodicTasks, IntervalSchedule, CrontabSchedule,
 )
@@ -425,3 +430,48 @@ def test_track_changes(self):
         y = PeriodicTasks.last_change()
         assert y
         assert y > x
+
+
+@pytest.mark.django_db()
+class test_modeladmin_PeriodicTaskAdmin(SchedulerCase):
+    @pytest.mark.django_db()
+    @pytest.fixture(autouse=True)
+    def setup_scheduler(self, app):
+        self.app = app
+        self.site = AdminSite()
+        self.request_factory = RequestFactory()
+
+        entry_name, entry = self.create_conf_entry()
+        self.app.conf.beat_schedule = {entry_name: entry}
+        self.m1 = PeriodicTask(name=entry_name)
+        self.m1.task = 'celery.backend_cleanup'
+        self.m1.save()
+
+        entry_name, entry = self.create_conf_entry()
+        self.app.conf.beat_schedule = {entry_name: entry}
+        self.m2 = PeriodicTask(name=entry_name)
+        self.m2.task = 'celery.backend_cleanup'
+        self.m2.save()
+
+    def patch_request(self, request):
+        """patch request to allow for django messages storage"""
+        setattr(request, 'session', 'session')
+        messages = FallbackStorage(request)
+        setattr(request, '_messages', messages)
+        return request
+
+    def test_run_task(self):
+        ma = PeriodicTaskAdmin(PeriodicTask, self.site)
+        self.request = self.patch_request(self.request_factory.get('/'))
+        ma.run_tasks(self.request, PeriodicTask.objects.filter(id=self.m1.id))
+        assert len(self.request._messages._queued_messages) == 1
+        queued_message = self.request._messages._queued_messages[0].message
+        assert queued_message == '1 task was successfully run'
+
+    def test_run_tasks(self):
+        ma = PeriodicTaskAdmin(PeriodicTask, self.site)
+        self.request = self.patch_request(self.request_factory.get('/'))
+        ma.run_tasks(self.request, PeriodicTask.objects.all())
+        assert len(self.request._messages._queued_messages) == 1
+        queued_message = self.request._messages._queued_messages[0].message
+        assert queued_message == '2 tasks were successfully run'