refactor: separated queue views to several files

Author: cunla

scheduler/helpers/queues/getters.py

@@ -12,7 +12,6 @@
 from .queue_logic import Queue
 
 
-
 _BAD_QUEUE_CONFIGURATION = set()
 
 
@@ -77,5 +76,3 @@ def get_all_workers() -> Set[WorkerModel]:
             logger.error(f"Could not connect for queue {queue_name}: {e}")
             _BAD_QUEUE_CONFIGURATION.add(queue_name)
     return workers_set
-
-

scheduler/management/commands/scheduler_stats.py

@@ -50,7 +50,7 @@ def add_arguments(self, parser):
     def _print_separator(self):
         click.echo("-" * self.table_width)
 
-    def _print_stats_dashboard(self, statistics, prev_stats=None, with_color:bool = True):
+    def _print_stats_dashboard(self, statistics, prev_stats=None, with_color: bool = True):
         if self.interval:
             click.clear()
         click.echo()

scheduler/templates/admin/scheduler/job_detail.html

@@ -149,27 +149,27 @@ <h2 class="fieldset-heading">Last result</h2>
         </fieldset>
         <div class="submit-row">
             <div class="deletelink-box">
-                <a href="{% url 'queue_job_action' job.name 'delete' %}" class="deletelink">Delete</a>
+                <a href="{% url 'job_detail_action' job.name 'delete' %}" class="deletelink">Delete</a>
             </div>
             {% if job.is_started %}
                 <div class="deletelink-box">
-                    <form method='POST' action="{% url 'queue_job_action' job.name 'cancel' %}">
+                    <form method='POST' action="{% url 'job_detail_action' job.name 'cancel' %}">
                         {% csrf_token %}
                         <input type="submit" value="cancel" class="default" name="cancel">
                     </form>
                 </div>
             {% endif %}
             {% if job.is_failed %}
                 <div class="deletelink-box">
-                    <form method='POST' action="{% url 'queue_job_action' job.name 'requeue' %}">
+                    <form method='POST' action="{% url 'job_detail_action' job.name 'requeue' %}">
                         {% csrf_token %}
                         <input type="submit" value="Requeue" class="default" name="requeue">
                     </form>
                 </div>
             {% endif %}
             {% if not job.is_queued and not job.is_failed %}
                 <div class="deletelink-box">
-                    <form method='POST' action="{% url 'queue_job_action' job.name 'enqueue' %}">
+                    <form method='POST' action="{% url 'job_detail_action' job.name 'enqueue' %}">
                         {% csrf_token %}
                         <input type="submit" value="Enqueue" class="default" name="Requeue">
                     </form>

scheduler/templates/admin/scheduler/jobs.html

@@ -31,7 +31,7 @@
             <li><a href="{% url 'queue_registry_action' queue.name registry_name 'empty' %}" class="deletelink">Empty Queue</a></li>
         </ul>
         <div class="module" id="changelist">
-            <form id="changelist-form" action="{% url 'queue_confirm_action' queue.name %}" method="post">
+            <form id="changelist-form" action="{% url 'queue_confirm_job_action' queue.name %}" method="post">
                 {% csrf_token %}
                 <div class="actions">
                     <label>Actions:

scheduler/tests/test_multiprocess/test_integrity.py

@@ -22,7 +22,7 @@ def test_cancel_job_after_it_started(self):
         job = JobModel.get(job.name, connection=queue.connection)
         self.assertEqual(JobStatus.STARTED, job.status)
         # act
-        res = self.client.post(reverse("queue_job_action", args=[job.name, "cancel"]), {"post": "yes"}, follow=True)
+        res = self.client.post(reverse("job_detail_action", args=[job.name, "cancel"]), {"post": "yes"}, follow=True)
 
         # assert
         self.assertEqual(200, res.status_code)

scheduler/tests/test_task_types/test_task_model.py

@@ -340,7 +340,7 @@ def test_admin_change_view__bad_redis_connection(self):
             # arrange
             self.client.login(username="admin", password="admin")
             task = task_factory(self.task_type, queue="test2", instance_only=True)
-            task.save(schedule_job=False)            
+            task.save(schedule_job=False)
             url = reverse(
                 f"admin:scheduler_task_change",
                 args=[

scheduler/tests/test_views/test_queue_job_action.py renamed to scheduler/tests/test_views/test_job_detail_action.py

@@ -11,7 +11,7 @@
 
 class SingleJobActionViewsTest(BaseTestCase):
     def test_single_job_action_unknown_job(self):
-        res = self.client.get(reverse("queue_job_action", args=["unknown", "cancel"]), follow=True)
+        res = self.client.get(reverse("job_detail_action", args=["unknown", "cancel"]), follow=True)
         self.assertEqual(200, res.status_code)
         assert_response_has_msg(res, "Job unknown does not exist, maybe its TTL has passed")
 
@@ -22,7 +22,7 @@ def test_single_job_action_unknown_action(self):
         worker.work()
         job = JobModel.get(job.name, connection=queue.connection)
         self.assertTrue(job.is_failed)
-        res = self.client.get(reverse("queue_job_action", args=[job.name, "unknown"]), follow=True)
+        res = self.client.get(reverse("job_detail_action", args=[job.name, "unknown"]), follow=True)
         self.assertEqual(400, res.status_code)
 
     def test_single_job_action_requeue_job(self):
@@ -32,18 +32,18 @@ def test_single_job_action_requeue_job(self):
         worker.work()
         job = JobModel.get(job.name, connection=queue.connection)
         self.assertTrue(job.is_failed)
-        res = self.client.get(reverse("queue_job_action", args=[job.name, "requeue"]), follow=True)
+        res = self.client.get(reverse("job_detail_action", args=[job.name, "requeue"]), follow=True)
         self.assertEqual(200, res.status_code)
-        self.client.post(reverse("queue_job_action", args=[job.name, "requeue"]), {"requeue": "Requeue"}, follow=True)
+        self.client.post(reverse("job_detail_action", args=[job.name, "requeue"]), {"requeue": "Requeue"}, follow=True)
         self.assertIn(job, JobModel.get_many(queue.queued_job_registry.all(), queue.connection))
         queue.delete_job(job.name)
 
     def test_single_job_action_delete_job(self):
         queue = get_queue("default")
         job = queue.create_and_enqueue_job(test_job, job_info_ttl=0)
-        res = self.client.get(reverse("queue_job_action", args=[job.name, "delete"]), follow=True)
+        res = self.client.get(reverse("job_detail_action", args=[job.name, "delete"]), follow=True)
         self.assertEqual(200, res.status_code)
-        self.client.post(reverse("queue_job_action", args=[job.name, "delete"]), {"post": "yes"}, follow=True)
+        self.client.post(reverse("job_detail_action", args=[job.name, "delete"]), {"post": "yes"}, follow=True)
         self.assertFalse(JobModel.exists(job.name, connection=queue.connection))
         self.assertNotIn(job.name, queue.queued_job_registry.all())
 
@@ -52,9 +52,9 @@ def test_single_job_action_cancel_job(self):
         job = queue.create_and_enqueue_job(long_job)
         self.assertTrue(job.is_queued)
         job = JobModel.get(job.name, connection=queue.connection)
-        res = self.client.get(reverse("queue_job_action", args=[job.name, "cancel"]), follow=True)
+        res = self.client.get(reverse("job_detail_action", args=[job.name, "cancel"]), follow=True)
         self.assertEqual(200, res.status_code)
-        res = self.client.post(reverse("queue_job_action", args=[job.name, "cancel"]), {"post": "yes"}, follow=True)
+        res = self.client.post(reverse("job_detail_action", args=[job.name, "cancel"]), {"post": "yes"}, follow=True)
         self.assertEqual(200, res.status_code)
         job = JobModel.get(job.name, connection=queue.connection)
         self.assertTrue(job.is_canceled)
@@ -63,12 +63,12 @@ def test_single_job_action_cancel_job(self):
     def test_single_job_action_cancel_job_that_is_already_cancelled(self):
         queue = get_queue("django_tasks_scheduler_test")
         job = queue.create_and_enqueue_job(long_job)
-        res = self.client.post(reverse("queue_job_action", args=[job.name, "cancel"]), {"post": "yes"}, follow=True)
+        res = self.client.post(reverse("job_detail_action", args=[job.name, "cancel"]), {"post": "yes"}, follow=True)
         self.assertEqual(200, res.status_code)
         tmp = JobModel.get(job.name, connection=queue.connection)
         self.assertTrue(tmp.is_canceled)
         self.assertNotIn(job.name, queue.queued_job_registry.all())
-        res = self.client.post(reverse("queue_job_action", args=[job.name, "cancel"]), {"post": "yes"}, follow=True)
+        res = self.client.post(reverse("job_detail_action", args=[job.name, "cancel"]), {"post": "yes"}, follow=True)
         self.assertEqual(200, res.status_code)
         assert_message_in_response(res, f"Could not perform action: Cannot cancel already canceled job: {job.name}")
 
@@ -86,9 +86,9 @@ def test_single_job_action_enqueue_job(self):
         self.assertIsNotNone(job_list[-1].enqueued_at)
 
         # Try to force enqueue last job should do nothing
-        res = self.client.get(reverse("queue_job_action", args=[job_list[-1].name, "enqueue"]), follow=True)
+        res = self.client.get(reverse("job_detail_action", args=[job_list[-1].name, "enqueue"]), follow=True)
         self.assertEqual(200, res.status_code)
-        res = self.client.post(reverse("queue_job_action", args=[job_list[-1].name, "enqueue"]), follow=True)
+        res = self.client.post(reverse("job_detail_action", args=[job_list[-1].name, "enqueue"]), follow=True)
 
         self.assertEqual(200, res.status_code)
         tmp = JobModel.get(job_list[-1].name, connection=queue.connection)
@@ -109,9 +109,9 @@ def test_single_job_action_enqueue_job_sync_queue(self):
         self.assertIsNotNone(job_list[-1].enqueued_at)
 
         # Try to force enqueue last job should do nothing
-        res = self.client.get(reverse("queue_job_action", args=[job_list[-1].name, "enqueue"]), follow=True)
+        res = self.client.get(reverse("job_detail_action", args=[job_list[-1].name, "enqueue"]), follow=True)
         self.assertEqual(200, res.status_code)
-        res = self.client.post(reverse("queue_job_action", args=[job_list[-1].name, "enqueue"]), follow=True)
+        res = self.client.post(reverse("job_detail_action", args=[job_list[-1].name, "enqueue"]), follow=True)
 
         self.assertEqual(200, res.status_code)
         tmp = JobModel.get(job_list[-1].name, connection=queue.connection)

scheduler/tests/test_views/test_job_details.py

@@ -114,8 +114,9 @@ def test_clear_queue_scheduled(self):
         self.assertEqual(200, res.status_code)
         self.assertEqual(res.context["jobs"], [job])
 
-        res = self.client.post(reverse("queue_registry_action", args=[queue.name, "scheduled", "empty"]),
-                               {"post": "yes"}, follow=True)
+        res = self.client.post(
+            reverse("queue_registry_action", args=[queue.name, "scheduled", "empty"]), {"post": "yes"}, follow=True
+        )
         assert_message_in_response(res, f"You have successfully cleared the scheduled jobs in queue {queue.name}")
         self.assertEqual(200, res.status_code)
         self.assertFalse(JobModel.exists(job.name, connection=queue.connection))

scheduler/tests/test_views/test_queue_actions.py

@@ -1,11 +1,11 @@
 from django.urls import reverse
 
 from scheduler.helpers.queues import get_queue
-from scheduler.worker import create_worker
 from scheduler.redis_models import JobStatus, JobModel
 from scheduler.tests.jobs import failing_job, test_job
 from scheduler.tests.test_views.base import BaseTestCase
 from scheduler.tests.testtools import assert_message_in_response
+from scheduler.worker import create_worker
 
 
 class QueueActionsViewsTest(BaseTestCase):
@@ -20,12 +20,7 @@ def test_job_list_action_delete_jobs__with_bad_next_url(self):
 
         # remove those jobs using view
         res = self.client.post(
-            reverse(
-                "queue_actions",
-                args=[
-                    queue.name,
-                ],
-            ),
+            reverse("queue_job_actions", args=[queue.name]),
             {
                 "action": "delete",
                 "job_names": job_names,
@@ -51,12 +46,7 @@ def test_job_list_action_delete_jobs(self):
 
         # remove those jobs using view
         res = self.client.post(
-            reverse(
-                "queue_actions",
-                args=[
-                    queue.name,
-                ],
-            ),
+            reverse("queue_job_actions", args=[queue.name]),
             {"action": "delete", "job_names": job_names},
             follow=True,
         )
@@ -87,7 +77,7 @@ def test_job_list_action_requeue_jobs(self):
             self.assertTrue(job.is_failed)
 
         # re-nqueue failed jobs from failed queue
-        self.client.post(reverse("queue_actions", args=[queue_name]), {"action": "requeue", "job_names": job_names})
+        self.client.post(reverse("queue_job_actions", args=[queue_name]), {"action": "requeue", "job_names": job_names})
 
         # check if we requeue all failed jobs
         for job_name in job_names:
@@ -115,7 +105,7 @@ def test_job_list_action_stop_jobs__move_to_finished_registry(self):
 
         # Stop those jobs using the view
         self.assertEqual(len(queue.active_job_registry), len(job_names))
-        self.client.post(reverse("queue_actions", args=[queue_name]), {"action": "stop", "job_names": job_names})
+        self.client.post(reverse("queue_job_actions", args=[queue_name]), {"action": "stop", "job_names": job_names})
         self.assertEqual(0, len(queue.active_job_registry))
 
         self.assertEqual(0, len(queue.canceled_job_registry))

scheduler/urls.py

@@ -7,14 +7,18 @@
     path("queues/stats.json", views.stats_json, name="queues_home_json"),
     path("queues/<str:queue_name>/workers/", views.queue_workers, name="queue_workers"),
     path("queues/<str:queue_name>/<str:registry_name>/jobs", views.list_registry_jobs, name="queue_registry_jobs"),
-    path("queues/<str:queue_name>/<str:registry_name>/<str:action>/", views.queue_registry_actions, name="queue_registry_action"),
-    path("queues/<str:queue_name>/confirm-action/", views.queue_confirm_action, name="queue_confirm_action"),
-    path("queues/<str:queue_name>/actions/", views.queue_job_actions, name="queue_actions"),
+    path(
+        "queues/<str:queue_name>/<str:registry_name>/<str:action>/",
+        views.queue_registry_actions,
+        name="queue_registry_action",
+    ),
+    path("queues/<str:queue_name>/confirm-action/", views.queue_confirm_job_action, name="queue_confirm_job_action"),
+    path("queues/<str:queue_name>/actions/", views.queue_job_actions, name="queue_job_actions"),
 ]
 
 urlpatterns += [
     path("workers/", views.workers_list, name="workers_home"),
     path("workers/<str:name>/", views.worker_details, name="worker_details"),
     path("jobs/<str:job_name>/", views.job_detail, name="job_details"),
-    path("jobs/<str:job_name>/<str:action>/", views.job_action, name="queue_job_action"),
+    path("jobs/<str:job_name>/<str:action>/", views.job_action, name="job_detail_action"),
 ]

scheduler/views/__init__.py

@@ -4,7 +4,7 @@
     "stats",
     "stats_json",
     "queue_registry_actions",
-    "queue_confirm_action",
+    "queue_confirm_job_action",
     "queue_workers",
     "queue_job_actions",
     "list_registry_jobs",
@@ -14,8 +14,8 @@
 ]
 
 from .job_views import job_detail, job_action
-from .queue_job_actions import queue_job_actions
-from .queue_registry_actions import queue_registry_actions, queue_confirm_action
+from .queue_job_actions import queue_job_actions, queue_confirm_job_action
+from .queue_registry_actions import queue_registry_actions
 from .queue_views import (
     stats,
     stats_json,

scheduler/views/queue_job_actions.py

@@ -1,9 +1,10 @@
+"""list_registry_jobs actions on multiple selected jobs"""
 from enum import Enum
 
-from django.contrib import messages
+from django.contrib import admin, messages
 from django.contrib.admin.views.decorators import staff_member_required
 from django.http import HttpResponse, HttpRequest
-from django.shortcuts import redirect
+from django.shortcuts import render, redirect
 from django.urls import reverse
 from django.views.decorators.cache import never_cache
 
@@ -26,11 +27,9 @@ def queue_job_actions(request: HttpRequest, queue_name: str) -> HttpResponse:
     next_url = _check_next_url(request, reverse("queue_registry_jobs", args=[queue_name, "queued"]))
     action = request.POST.get("action", False)
     job_names = request.POST.get("job_names", False)
-    if request.method != "POST" or not action or not job_names:
+    if request.method != "POST" or not action or not job_names or action not in QueueJobAction:
         return redirect(next_url)
     job_names = request.POST.getlist("job_names")
-    if action not in QueueJobAction:
-        return redirect(next_url)
     if action == QueueJobAction.DELETE:
         jobs = JobModel.get_many(job_names, connection=queue.connection)
         for job in jobs:
@@ -57,3 +56,31 @@ def queue_job_actions(request: HttpRequest, queue_name: str) -> HttpResponse:
                 pass
         messages.info(request, f"You have successfully stopped {cancelled_jobs}  jobs!")
     return redirect(next_url)
+
+
+@never_cache
+@staff_member_required
+def queue_confirm_job_action(request: HttpRequest, queue_name: str) -> HttpResponse:
+    queue = get_queue(queue_name)
+    next_url = _check_next_url(request, reverse("queue_registry_jobs", args=[queue_name, "queued"]))
+    action = request.POST.get("action", None)
+    job_names = request.POST.getlist("_selected_action", None)
+    if request.method != "POST" or action is None or job_names is None or action not in QueueJobAction:
+        return redirect(next_url)
+
+    # confirm action
+    context_data = {
+        **admin.site.each_context(request),
+        "action": action,
+        "jobs": [JobModel.get(job_name, connection=queue.connection) for job_name in job_names],
+        "total_jobs": len(job_names),
+        "queue": queue,
+        "next_url": next_url,
+        "action_url": reverse(
+            "queue_job_actions",
+            args=[
+                queue_name,
+            ],
+        ),
+    }
+    return render(request, "admin/scheduler/confirm_action.html", context_data)

scheduler/views/queue_registry_actions.py

@@ -1,3 +1,4 @@
+"""list_registry_jobs actions on all jobs in the registry"""
 from enum import Enum
 
 from django.contrib import admin, messages
@@ -66,43 +67,6 @@ def queue_registry_actions(request: HttpRequest, queue_name: str, registry_name:
         "action": action,
         "jobs": job_list,
         "next_url": next_url,
-        "action_url": reverse(
-            "queue_registry_action",
-            args=[
-                queue_name,
-                registry_name,
-                action
-            ],
-        ),
-    }
-    return render(request, "admin/scheduler/confirm_action.html", context_data)
-
-
-@never_cache
-@staff_member_required
-def queue_confirm_action(request: HttpRequest, queue_name: str) -> HttpResponse:
-    queue = get_queue(queue_name)
-    next_url = _check_next_url(request, reverse("queue_registry_jobs", args=[queue_name, "queued"]))
-    if request.method != "POST":
-        return redirect(next_url)
-    action = request.POST.get("action", None)
-    job_names = request.POST.getlist("_selected_action", None)
-    if action is None or job_names is None:
-        return redirect(next_url)
-
-    # confirm action
-    context_data = {
-        **admin.site.each_context(request),
-        "action": action,
-        "jobs": [JobModel.get(job_name, connection=queue.connection) for job_name in job_names],
-        "total_jobs": len(job_names),
-        "queue": queue,
-        "next_url": next_url,
-        "action_url": reverse(
-            "queue_actions",
-            args=[
-                queue_name,
-            ],
-        ),
+        "action_url": reverse("queue_registry_action", args=[queue_name, registry_name, action]),
     }
     return render(request, "admin/scheduler/confirm_action.html", context_data)