refactor: separated queue views to several files

Author: cunla

scheduler/helpers/callback.py

@@ -1,7 +1,7 @@
-import importlib
 import inspect
 from typing import Union, Callable, Any, Optional
 
+from scheduler.helpers.utils import callable_func
 from scheduler.timeouts import JobTimeoutException
 
 
@@ -19,7 +19,11 @@ def __init__(self, func: Union[str, Callable[..., Any]], timeout: Optional[int]
         if not isinstance(func, str) and not inspect.isfunction(func) and not inspect.isbuiltin(func):
             raise CallbackSetupError(f"Callback `func` must be a string or function, received {func}")
         if isinstance(func, str):
-            func = _import_attribute(func)
+            try:
+                func_str = func
+                func = callable_func(func)
+            except (TypeError, AttributeError, ModuleNotFoundError, ValueError) as e:
+                raise CallbackSetupError(f"Callback `func` is not callable: {func_str}")
         self.func: Callable[..., Any] = func
 
     @property
@@ -31,47 +35,3 @@ def __call__(self, *args, **kwargs):
 
         with SCHEDULER_CONFIG.DEATH_PENALTY_CLASS(self.timeout, JobTimeoutException):
             return self.func(*args, **kwargs)
-
-
-def _import_attribute(name: str) -> Callable[..., Any]:
-    """Returns an attribute from a dotted path name. Example: `path.to.func`.
-
-    When the attribute we look for is a staticmethod, module name in its dotted path is not the last-before-end word
-    E.g.: package_a.package_b.module_a.ClassA.my_static_method
-    Thus we remove the bits from the end of the name until we can import it
-
-    :param name: The name (reference) to the path.
-    :raises ValueError: If no module is found or invalid attribute name.
-    :returns: An attribute (normally a Callable)
-    """
-    name_bits = name.split(".")
-    module_name_bits, attribute_bits = name_bits[:-1], [name_bits[-1]]
-    module = None
-    while len(module_name_bits) > 0:
-        try:
-            module_name = ".".join(module_name_bits)
-            module = importlib.import_module(module_name)
-            break
-        except ImportError:
-            attribute_bits.insert(0, module_name_bits.pop())
-
-    if module is None:  # maybe it's a builtin
-        try:
-            return __builtins__[name]
-        except KeyError:
-            raise CallbackSetupError(f"Invalid attribute name: {name}")
-
-    attribute_name = ".".join(attribute_bits)
-    if hasattr(module, attribute_name):
-        return getattr(module, attribute_name)
-    # staticmethods
-    attribute_name = attribute_bits.pop()
-    attribute_owner_name = ".".join(attribute_bits)
-    try:
-        attribute_owner = getattr(module, attribute_owner_name)
-    except:  # noqa
-        raise CallbackSetupError(f"Invalid attribute name: {attribute_name}")
-
-    if not hasattr(attribute_owner, attribute_name):
-        raise CallbackSetupError(f"Invalid attribute name: {name}")
-    return getattr(attribute_owner, attribute_name)

scheduler/helpers/queues/queue_logic.py

@@ -2,7 +2,6 @@
 import sys
 import traceback
 from datetime import datetime
-from functools import total_ordering
 from typing import Dict, List, Optional, Tuple, Union, Self, Any
 
 from redis import WatchError
@@ -56,7 +55,6 @@ def perform_job(job_model: JobModel, connection: ConnectionType) -> Any:  # noqa
 _job_stack = []
 
 
-@total_ordering
 class Queue:
     REGISTRIES = dict(
         finished="finished_job_registry",
@@ -146,17 +144,6 @@ def clean_registries(self, timestamp: Optional[float] = None) -> None:
     def first_queued_job_name(self) -> Optional[str]:
         return self.queued_job_registry.get_first()
 
-    def empty(self):
-        """Removes all queued jobs from the queue."""
-        queued_jobs_count = self.queued_job_registry.count(connection=self.connection)
-        with self.connection.pipeline() as pipe:
-            for offset in range(0, queued_jobs_count, 1000):
-                job_names = self.queued_job_registry.all(offset, 1000)
-                for job_name in job_names:
-                    self.queued_job_registry.delete(connection=pipe, job_name=job_name)
-                JobModel.delete_many(job_names, connection=pipe)
-            pipe.execute()
-
     @property
     def count(self) -> int:
         """Returns a count of all messages in the queue."""
@@ -186,24 +173,24 @@ def get_all_jobs(self) -> List[JobModel]:
         return JobModel.get_many(job_names, connection=self.connection)
 
     def create_and_enqueue_job(
-        self,
-        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,
-        description: Optional[str] = None,
-        name: Optional[str] = None,
-        at_front: bool = False,
-        meta: Optional[Dict] = None,
-        on_success: Optional[Callback] = None,
-        on_failure: Optional[Callback] = None,
-        on_stopped: Optional[Callback] = None,
-        task_type: Optional[str] = None,
-        scheduled_task_id: Optional[int] = None,
-        pipeline: Optional[ConnectionType] = None,
+            self,
+            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,
+            description: Optional[str] = None,
+            name: Optional[str] = None,
+            at_front: bool = False,
+            meta: Optional[Dict] = None,
+            on_success: Optional[Callback] = None,
+            on_failure: Optional[Callback] = None,
+            on_stopped: Optional[Callback] = None,
+            task_type: Optional[str] = None,
+            scheduled_task_id: Optional[int] = None,
+            pipeline: Optional[ConnectionType] = None,
     ) -> JobModel:
         """Creates a job to represent the delayed function call and enqueues it.
         :param when: When to schedule the job (None to enqueue immediately)
@@ -309,43 +296,6 @@ def run_job(self, job: JobModel) -> JobModel:
                 pipeline.execute()
         return job
 
-    def enqueue_job(
-        self, job_model: JobModel, connection: Optional[ConnectionType] = None, at_front: bool = False
-    ) -> JobModel:
-        """Enqueues a job for delayed execution without checking dependencies.
-
-        If Queue is instantiated with is_async=False, job is executed immediately.
-        :param job_model: The job redis model
-        :param connection: The Redis Pipeline
-        :param at_front: Whether to enqueue the job at the front
-
-        :returns: The enqueued JobModel
-        """
-
-        pipe = connection if connection is not None else self.connection.pipeline()
-
-        # Add Queue key set
-        job_model.status = JobStatus.QUEUED
-        job_model.enqueued_at = utcnow()
-        job_model.save(connection=pipe)
-
-        if self._is_async:
-            if at_front:
-                score = current_timestamp()
-            else:
-                score = self.queued_job_registry.get_last_timestamp() or current_timestamp()
-            self.scheduled_job_registry.delete(connection=pipe, job_name=job_model.name)
-            self.queued_job_registry.add(connection=pipe, score=score, job_name=job_model.name)
-            pipe.execute()
-            logger.debug(f"Pushed job {job_model.name} into {self.name} queued-jobs registry")
-        else:  # sync mode
-            pipe.execute()
-            job_model = self.run_sync(job_model)
-            job_model.expire(ttl=job_model.job_info_ttl, connection=pipe)
-            pipe.execute()
-
-        return job_model
-
     def run_sync(self, job: JobModel) -> JobModel:
         """Run a job synchronously, meaning on the same process the method was called."""
         job.prepare_for_execution("sync", self.active_job_registry, self.connection)
@@ -361,10 +311,10 @@ def run_sync(self, job: JobModel) -> JobModel:
 
     @classmethod
     def dequeue_any(
-        cls,
-        queues: List[Self],
-        timeout: Optional[int],
-        connection: Optional[ConnectionType] = None,
+            cls,
+            queues: List[Self],
+            timeout: Optional[int],
+            connection: Optional[ConnectionType] = None,
     ) -> Tuple[Optional[JobModel], Optional[Self]]:
         """Class method returning a Job instance at the front of the given set of Queues, where the order of the queues
         is important.
@@ -398,19 +348,6 @@ def dequeue_any(
             return job, queue
         return None, None
 
-    def __eq__(self, other: Self) -> bool:
-        if not isinstance(other, Queue):
-            raise TypeError("Cannot compare queues to other objects")
-        return self.name == other.name
-
-    def __lt__(self, other: Self) -> bool:
-        if not isinstance(other, Queue):
-            raise TypeError("Cannot compare queues to other objects")
-        return self.name < other.name
-
-    def __hash__(self) -> int:
-        return hash(self.name)
-
     def __repr__(self) -> str:
         return f"{self.__class__.__name__}({self.name!r})"
 
@@ -479,17 +416,39 @@ def delete_job(self, job_name: str, expire_job_model: bool = True) -> None:
             except WatchError:
                 pass
 
-    def requeue_jobs(self, *job_names: str, at_front: bool = False) -> int:
-        jobs = JobModel.get_many(job_names, connection=self.connection)
-        jobs_requeued = 0
-        with self.connection.pipeline() as pipe:
-            for job in jobs:
-                if job is None:
-                    continue
-                job.started_at = None
-                job.ended_at = None
-                job.save(connection=pipe)
-                self.enqueue_job(job, connection=pipe, at_front=at_front)
-                jobs_requeued += 1
+    def enqueue_job(
+            self, job_model: JobModel, connection: Optional[ConnectionType] = None, at_front: bool = False
+    ) -> JobModel:
+        """Enqueues a job for delayed execution without checking dependencies.
+
+        If Queue is instantiated with is_async=False, job is executed immediately.
+        :param job_model: The job redis model
+        :param connection: The Redis Pipeline
+        :param at_front: Whether to enqueue the job at the front
+
+        :returns: The enqueued JobModel
+        """
+
+        pipe = connection if connection is not None else self.connection.pipeline()
+        job_model.started_at = None
+        job_model.ended_at = None
+        job_model.status = JobStatus.QUEUED
+        job_model.enqueued_at = utcnow()
+        job_model.save(connection=pipe)
+
+        if self._is_async:
+            if at_front:
+                score = current_timestamp()
+            else:
+                score = self.queued_job_registry.get_last_timestamp() or current_timestamp()
+            self.scheduled_job_registry.delete(connection=pipe, job_name=job_model.name)
+            self.queued_job_registry.add(connection=pipe, score=score, job_name=job_model.name)
             pipe.execute()
-        return jobs_requeued
+            logger.debug(f"Pushed job {job_model.name} into {self.name} queued-jobs registry")
+        else:  # sync mode
+            pipe.execute()
+            job_model = self.run_sync(job_model)
+            job_model.expire(ttl=job_model.job_info_ttl, connection=pipe)
+            pipe.execute()
+
+        return job_model

scheduler/redis_models/registry/queue_registries.py

@@ -28,6 +28,15 @@ def compact(self) -> None:
             if not JobModel.exists(job_name, self.connection):
                 self.delete(connection=self.connection, job_name=job_name)
 
+    def empty(self) -> None:
+        queued_jobs_count = self.count(connection=self.connection)
+        with self.connection.pipeline() as pipe:
+            for offset in range(0, queued_jobs_count, 1000):
+                job_names = self.all(offset, 1000)
+                for job_name in job_names:
+                    self.delete(connection=pipe, job_name=job_name)
+                JobModel.delete_many(job_names, connection=pipe)
+            pipe.execute()
 
 class FinishedJobRegistry(JobNamesRegistry):
     _element_key_template: ClassVar[str] = ":registry:{}:finished_jobs"

scheduler/tests/test_internals.py

@@ -29,13 +29,13 @@ def test_callback_bad_arguments(self):
         self.assertEqual(str(cm.exception), "Callback `timeout` must be a positive int, but received 1m")
         with self.assertRaises(CallbackSetupError) as cm:
             Callback("scheduler.tests.jobs.non_existing_method")
-        self.assertEqual(str(cm.exception), "Invalid attribute name: non_existing_method")
+        self.assertEqual(str(cm.exception), "Callback `func` is not callable: scheduler.tests.jobs.non_existing_method")
         with self.assertRaises(CallbackSetupError) as cm:
             Callback("scheduler.tests.non_existing_module.non_existing_method")
-        self.assertEqual(str(cm.exception), "Invalid attribute name: non_existing_method")
+        self.assertEqual(str(cm.exception), "Callback `func` is not callable: scheduler.tests.non_existing_module.non_existing_method")
         with self.assertRaises(CallbackSetupError) as cm:
             Callback("non_existing_method")
-        self.assertEqual(str(cm.exception), "Invalid attribute name: non_existing_method")
+        self.assertEqual(str(cm.exception), "Callback `func` is not callable: non_existing_method")
         with self.assertRaises(CallbackSetupError) as cm:
             Callback(1)
         self.assertEqual(str(cm.exception), "Callback `func` must be a string or function, received 1")

scheduler/tests/test_mgmt_commands/test_run_job.py

@@ -10,7 +10,7 @@
 class RunJobTest(TestCase):
     def test_run_job__should_schedule_job(self):
         queue = get_queue("default")
-        queue.empty()
+        queue.queued_job_registry.empty()
         func_name = f"{test_job.__module__}.{test_job.__name__}"
         # act
         call_command("run_job", func_name, queue="default")

scheduler/tests/test_views/test_job_detail_action.py

@@ -25,19 +25,6 @@ def test_single_job_action_unknown_action(self):
         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):
-        queue = get_queue("default")
-        job = queue.create_and_enqueue_job(failing_job)
-        worker = create_worker("default", burst=True)
-        worker.work()
-        job = JobModel.get(job.name, connection=queue.connection)
-        self.assertTrue(job.is_failed)
-        res = self.client.get(reverse("job_detail_action", args=[job.name, "requeue"]), follow=True)
-        self.assertEqual(200, res.status_code)
-        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)
@@ -80,8 +67,6 @@ def test_single_job_action_enqueue_job(self):
             job = queue.create_and_enqueue_job(test_job)
             job_list.append(job)
 
-        # This job is deferred
-
         self.assertEqual(job_list[-1].get_status(connection=queue.connection), JobStatus.QUEUED)
         self.assertIsNotNone(job_list[-1].enqueued_at)
 

scheduler/tests/testtools.py

@@ -11,12 +11,11 @@
 from scheduler import settings
 from scheduler.admin.task_admin import job_execution_of
 from scheduler.helpers.queues import get_queue
-from scheduler.settings import QueueNotFoundError
-from scheduler.worker import create_worker
-from scheduler.models import TaskKwarg
 from scheduler.models import Task, TaskType
+from scheduler.models import TaskKwarg
 from scheduler.redis_models import JobModel
 from scheduler.worker import Worker
+from scheduler.worker import create_worker
 
 multiprocessing.set_start_method("fork")
 
@@ -48,7 +47,7 @@ def sequence_gen():
 
 
 def task_factory(
-    task_type: TaskType, callable_name: str = "scheduler.tests.jobs.test_job", instance_only=False, **kwargs
+        task_type: TaskType, callable_name: str = "scheduler.tests.jobs.test_job", instance_only=False, **kwargs
 ):
     values = dict(
         name="Scheduled Job %d" % next(seq),
@@ -135,8 +134,3 @@ def setUp(self) -> None:
 
     def tearDown(self) -> None:
         super(SchedulerBaseCase, self).tearDown()
-        try:
-            queue = get_queue("default")
-            queue.empty()
-        except QueueNotFoundError:
-            pass