refactor: separated queue views to several files

Author: cunla

scheduler/decorators.py

@@ -90,6 +90,7 @@ def delay(*args, **kwargs):
                 on_failure=self.on_failure,
                 on_success=self.on_success,
                 on_stopped=self.on_stopped,
+                when=None,
             )
 
         JOB_METHODS_LIST.append(f"{f.__module__}.{f.__name__}")

scheduler/helpers/queues/__init__.py

@@ -3,9 +3,8 @@
     "InvalidJobOperation",
     "get_queue",
     "get_all_workers",
-    "get_queues",
     "perform_job",
 ]
 
-from .getters import get_queue, get_all_workers, get_queues
+from .getters import get_queue, get_all_workers
 from .queue_logic import Queue, InvalidJobOperation, perform_job

scheduler/helpers/queues/getters.py

@@ -1,6 +1,5 @@
 from typing import List, Set
 
-from scheduler.types import ConnectionErrorTypes, BrokerMetaData, Broker
 from scheduler.redis_models.worker import WorkerModel
 from scheduler.settings import (
     SCHEDULER_CONFIG,
@@ -9,12 +8,10 @@
     QueueConfiguration,
     logger,
 )
+from scheduler.types import ConnectionErrorTypes, BrokerMetaData, Broker
 from .queue_logic import Queue
 
 
-class QueueConnectionDiscrepancyError(Exception):
-    pass
-
 
 _BAD_QUEUE_CONFIGURATION = set()
 
@@ -82,19 +79,3 @@ def get_all_workers() -> Set[WorkerModel]:
     return workers_set
 
 
-def get_queues(*queue_names: str) -> List[Queue]:
-    """Return queue instances from specified queue names. All instances must use the same Broker configuration."""
-
-    queue_config = get_queue_configuration(queue_names[0])
-    queues = [get_queue(queue_names[0])]
-    # perform consistency checks while building return list
-    for queue_name in queue_names[1:]:
-        curr_queue_config = get_queue_configuration(queue_name)
-        if not queue_config.same_connection_params(curr_queue_config):
-            raise QueueConnectionDiscrepancyError(
-                f'Queues must have the same broker connection. "{queue_name}" and "{queue_names[0]}" have different connection settings'
-            )
-        queue = get_queue(queue_name)
-        queues.append(queue)
-
-    return queues

scheduler/helpers/queues/queue_logic.py

@@ -190,6 +190,7 @@ def create_and_enqueue_job(
         func: FunctionReferenceType,
         args: Union[Tuple, List, None] = None,
         kwargs: Optional[Dict] = None,
+        when: Optional[datetime] = None,
         timeout: Optional[int] = None,
         result_ttl: Optional[int] = None,
         job_info_ttl: Optional[int] = None,
@@ -202,7 +203,6 @@ def create_and_enqueue_job(
         on_stopped: Optional[Callback] = None,
         task_type: Optional[str] = None,
         scheduled_task_id: Optional[int] = None,
-        when: Optional[datetime] = None,
         pipeline: Optional[ConnectionType] = None,
     ) -> JobModel:
         """Creates a job to represent the delayed function call and enqueues it.

scheduler/management/commands/run_job.py

@@ -32,6 +32,6 @@ def handle(self, **options):
         queue = get_queue(options.get("queue"))
         func = options.get("callable")
         args = options.get("args")
-        job = queue.create_and_enqueue_job(func, args=args, timeout=timeout, result_ttl=result_ttl)
+        job = queue.create_and_enqueue_job(func, args=args, timeout=timeout, result_ttl=result_ttl, when=None)
         if verbosity:
             click.echo(f"Job {job.name} created")

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):
+    def _print_stats_dashboard(self, statistics, prev_stats=None, with_color:bool = True):
         if self.interval:
             click.clear()
         click.echo()
@@ -62,6 +62,8 @@ def _print_stats_dashboard(self, statistics, prev_stats=None):
         for ind, queue in enumerate(statistics["queues"]):
             vals = list((queue[k] for k in KEYS))
             # Deal with colors
+            if not with_color:
+                colors = ["" for _ in KEYS]
             if prev_stats and len(prev_stats["queues"]) > ind:
                 prev = prev_stats["queues"][ind]
                 prev_vals = tuple(prev[k] for k in KEYS)
@@ -71,7 +73,7 @@ def _print_stats_dashboard(self, statistics, prev_stats=None):
             else:
                 colors = [ANSI_LIGHT_WHITE for _ in range(len(vals))]
             to_print = " | ".join([f"{colors[i]}{vals[i]:9}{ANSI_RESET}" for i in range(len(vals))])
-            click.echo(f"| {queue['name']:<16} | {to_print} |", color=True)
+            click.echo(f"| {queue['name']:<16} | {to_print} |", color=with_color)
 
         self._print_separator()
 
@@ -98,22 +100,20 @@ def handle(self, *args, **options):
                 click.secho("Aborting. yaml not supported", err=True, fg="red")
                 return
 
-            click.secho(
-                yaml.dump(get_statistics(), default_flow_style=False),
-            )
+            click.secho(yaml.dump(get_statistics(), default_flow_style=False))
             return
 
         self.interval = options.get("interval")
 
         if not self.interval or self.interval < 0:
-            self._print_stats_dashboard(get_statistics())
+            self._print_stats_dashboard(get_statistics(), with_color=not options.get("no_color"))
             return
 
         try:
             prev = None
             while True:
                 statistics = get_statistics()
-                self._print_stats_dashboard(statistics, prev)
+                self._print_stats_dashboard(statistics, prev, with_color=not options.get("no_color"))
                 prev = statistics
                 time.sleep(self.interval)
         except KeyboardInterrupt:

scheduler/models/task.py

@@ -15,12 +15,12 @@
 from django.utils.safestring import mark_safe
 from django.utils.translation import gettext_lazy as _
 
+from scheduler import settings
 from scheduler.helpers.callback import Callback
 from scheduler.helpers.queues import Queue
 from scheduler.helpers.queues import get_queue
 from scheduler.redis_models import JobModel
 from scheduler.settings import logger, get_queue_names
-from scheduler import settings
 from scheduler.types import ConnectionType, TASK_TYPES
 from .args import TaskArg, TaskKwarg
 from ..helpers import utils
@@ -255,11 +255,7 @@ def rqueue(self) -> Queue:
     def enqueue_to_run(self) -> bool:
         """Enqueue task to run now as a different instance from the scheduled task."""
         kwargs = self._enqueue_args()
-        self.rqueue.create_and_enqueue_job(
-            run_task,
-            args=(self.task_type, self.id),
-            **kwargs,
-        )
+        self.rqueue.create_and_enqueue_job(run_task, args=(self.task_type, self.id), when=None, **kwargs)
         return True
 
     def unschedule(self) -> bool:

scheduler/redis_models/registry/queue_registries.py

@@ -54,7 +54,7 @@ def cleanup(self, connection: ConnectionType, timestamp: Optional[float] = None)
         and `all()` methods implemented in JobIdsRegistry."""
         pass
 
-    def schedule(self, connection: ConnectionType, job: JobModel, scheduled_datetime):
+    def schedule(self, connection: ConnectionType, job: JobModel, scheduled_datetime: datetime):
         """
         Adds job to registry, scored by its execution time (in UTC).
         If datetime has no tzinfo, it will assume localtimezone.

scheduler/templates/admin/scheduler/jobs.html

@@ -25,10 +25,10 @@
     <div id="content-main">
         <ul class="object-tools">
             {% if job_status == 'Failed' %}
-                <li><a href="{% url 'queue_requeue_all' queue.name registry_name %}" class="requeuelink">Requeue All</a>
+                <li><a href="{% url 'queue_registry_action' queue.name registry_name 'requeue' %}" class="requeuelink">Requeue All</a>
                 </li>
             {% endif %}
-            <li><a href="{% url 'queue_clear' queue.name registry_name %}" class="deletelink">Empty Queue</a></li>
+            <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">

scheduler/tests/test_mgmt_commands/test_scheduler_stats.py

@@ -1,13 +1,94 @@
+import json
+import sys
+from io import StringIO
+
+import yaml
 from django.core.management import call_command
 from django.test import TestCase, override_settings
 
-from scheduler.tests import test_settings  # noqa
-from scheduler.types import SchedulerConfiguration
+from scheduler import settings
+from scheduler.helpers.queues import get_queue
+from scheduler.worker.worker import get_queues
 
 
+@override_settings(SCHEDULER_QUEUES=dict(default={"HOST": "localhost", "PORT": 6379, "DB": 0}))
 class SchedulerStatsTest(TestCase):
-    @override_settings(SCHEDULER_CONFIG=SchedulerConfiguration(SCHEDULER_INTERVAL=1))
-    def test_scheduler_stats__does_not_fail(self):
+    EXPECTED_OUTPUT = {
+        "queues": [
+            {
+                "canceled_jobs": 0,
+                "failed_jobs": 0,
+                "finished_jobs": 0,
+                "name": "default",
+                "oldest_job_timestamp": None,
+                "queued_jobs": 0,
+                "scheduled_jobs": 0,
+                "scheduler_pid": None,
+                "started_jobs": 0,
+                "workers": 0,
+            }
+        ]
+    }
+    OLD_QUEUES = None
+
+    def setUp(self):
+        super(SchedulerStatsTest, self).setUp()
+        SchedulerStatsTest.OLD_QUEUES = settings._QUEUES
+        settings._QUEUES = dict()
+        settings.conf_settings()
+        get_queue("default").connection.flushall()
+
+    def tearDown(self):
+        super(SchedulerStatsTest, self).tearDown()
+        settings._QUEUES = SchedulerStatsTest.OLD_QUEUES
+
+    def test_scheduler_stats__json_output(self):
+        test_stdout = StringIO()
+        sys.stdout = test_stdout
+        # act
         call_command("scheduler_stats", "-j")
+        # assert
+        res = test_stdout.getvalue()
+        self.assertEqual(json.loads(res), SchedulerStatsTest.EXPECTED_OUTPUT)
+
+    def test_scheduler_stats__yaml_output(self):
+        # arrange
+        test_stdout = StringIO()
+        sys.stdout = test_stdout
+        # act
         call_command("scheduler_stats", "-y")
-        call_command("scheduler_stats")
+        # assert
+        res = test_stdout.getvalue()
+        self.assertEqual(yaml.load(res, yaml.SafeLoader), SchedulerStatsTest.EXPECTED_OUTPUT)
+
+    def test_scheduler_stats__plain_text_output(self):
+        test_stdout = StringIO()
+        sys.stdout = test_stdout
+        # act
+        call_command("scheduler_stats", "--no-color")
+        # assert
+        res = test_stdout.getvalue()
+        self.assertEqual(
+            res,
+            """
+Django-Scheduler CLI Dashboard
+
+--------------------------------------------------------------------------------
+| Name             |    Queued |    Active |  Finished |  Canceled |   Workers |
+--------------------------------------------------------------------------------
+| default          |         0 |         0 |         0 |         0 |         0 |
+--------------------------------------------------------------------------------
+""",
+        )
+
+    def test_scheduler_stats__bad_args(self):
+        # arrange
+        sys.stderr = StringIO()
+        sys.stdout = StringIO()
+        # act
+        call_command("scheduler_stats", "-y", "-j")
+        # assert
+        res = sys.stdout.getvalue()
+        self.assertEqual(res, """""")
+        err = sys.stderr.getvalue()
+        self.assertEqual(err, """Aborting. Cannot output as both json and yaml\n""")