Remove RQ dependency

cunla · cunla · commit 78b5c2ab26c3 · 2025-03-24T10:16:35.000-04:00
diff --git a/scheduler/helpers/callback.py b/scheduler/helpers/callback.py
@@ -47,20 +47,19 @@ def _import_attribute(name: str) -> Callable[..., Any]:
     name_bits = name.split(".")
     module_name_bits, attribute_bits = name_bits[:-1], [name_bits[-1]]
     module = None
-    while len(module_name_bits):
+    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
+    if module is None: # maybe it's a builtin
         try:
             return __builtins__[name]
         except KeyError:
-            raise ValueError(f"Invalid attribute name: {name}")
+            raise CallbackSetupError(f"Invalid attribute name: {name}")
 
     attribute_name = ".".join(attribute_bits)
     if hasattr(module, attribute_name):
diff --git a/scheduler/helpers/queues/queue_logic.py b/scheduler/helpers/queues/queue_logic.py
@@ -1,7 +1,7 @@
 import asyncio
 import sys
 import traceback
-from datetime import datetime, timedelta
+from datetime import datetime
 from functools import total_ordering
 from typing import Dict, List, Optional, Tuple, Union, Self, Any
 
@@ -317,7 +317,6 @@ def enqueue_job(
         job_model.status = JobStatus.QUEUED
         job_model.enqueued_at = utcnow()
         job_model.save(connection=pipe)
-        job_model.expire(ttl=job_model.job_info_ttl, connection=pipe)
 
         if self._is_async:
             if at_front:
@@ -326,10 +325,13 @@ def enqueue_job(
                 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)
-            result = pipe.execute()
-            logger.debug(f"Pushed job {job_model.name} into {self.name}, {result[3]} job(s) are in queue.")
+            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
 
@@ -446,17 +448,18 @@ def cancel_job(self, job_name: str) -> None:
                 # caller to handle it
                 raise
 
-    def delete_job(self, job_name: str):
+    def delete_job(self, job_name: str, expire_job_model: bool = True) -> None:
         """Deletes the given job from the queue and all its registries"""
-
         pipe = self.connection.pipeline()
 
         while True:
             try:
                 self._remove_from_registries(job_name, connection=pipe)
                 self.failed_job_registry.delete(connection=pipe, job_name=job_name)
-                if JobModel.exists(job_name, connection=self.connection):
-                    JobModel.delete_many([job_name], connection=pipe)
+                if expire_job_model:
+                    job_model = JobModel.get(job_name, connection=self.connection)
+                    if job_model is not None:
+                        job_model.expire(ttl=job_model.job_info_ttl, connection=pipe)
                 pipe.execute()
                 break
             except WatchError:
diff --git a/scheduler/models/task.py b/scheduler/models/task.py
@@ -351,9 +351,9 @@ def _schedule(self) -> bool:
         if not self.enabled:
             logger.debug(f"Task {str(self)} disabled, enable task before scheduling")
             return False
-        if self.task_type in {TaskType.REPEATABLE, TaskType.ONCE} and self._schedule_time() < timezone.now():
-            return False
         schedule_time = self._schedule_time()
+        if self.task_type in {TaskType.REPEATABLE, TaskType.ONCE} and schedule_time < timezone.now():
+            return False
         kwargs = self._enqueue_args()
         job = self.rqueue.create_and_enqueue_job(
             tools.run_task,
diff --git a/scheduler/redis_models/base.py b/scheduler/redis_models/base.py
@@ -102,7 +102,7 @@ def deserialize(cls, data: Dict[str, Any]) -> Self:
         types = {f.name: f.type for f in dataclasses.fields(cls)}
         for k in data:
             if k not in types:
-                logger.warning(f"Unknown field {k} in WorkerModel")
+                logger.warning(f"Unknown field {k} in {cls.__name__}")
                 continue
             data[k] = _deserialize(data[k], types[k])
         return cls(**data)
@@ -184,6 +184,7 @@ def save(self, connection: ConnectionType) -> None:
             connection.hdel(self._key, *none_values)
         mapping = {k: v for k, v in mapping.items() if v is not None}
         if mapping:
+            logger.debug(f"Saving {self._key}")
             connection.hset(self._key, mapping=mapping)
         self._dirty_fields = set()
         self._save_all = False
diff --git a/scheduler/redis_models/job.py b/scheduler/redis_models/job.py
@@ -196,9 +196,7 @@ def create(
         timeout = _parse_timeout(timeout) or SCHEDULER_CONFIG.DEFAULT_JOB_TIMEOUT
         if timeout == 0:
             raise ValueError("0 timeout is not allowed. Use -1 for infinite timeout")
-        job_info_ttl = _parse_timeout(job_info_ttl or SCHEDULER_CONFIG.DEFAULT_JOB_TTL)
-        if job_info_ttl is not None and job_info_ttl <= 0:
-            raise ValueError("Job info ttl must be greater than 0")
+        job_info_ttl = _parse_timeout(job_info_ttl if job_info_ttl is not None else SCHEDULER_CONFIG.DEFAULT_JOB_TTL)
         result_ttl = _parse_timeout(result_ttl)
         if not isinstance(args, (tuple, list)):
             raise TypeError(f"{args!r} is not a valid args list")
@@ -228,7 +226,7 @@ def create(
         else:
             raise TypeError(f"Expected a callable or a string, but got: {func}")
         description = description or _get_call_string(func, args or [], kwargs or {}, max_length=75)
-
+        job_info_ttl = job_info_ttl if job_info_ttl is not None else SCHEDULER_CONFIG.DEFAULT_JOB_TTL
         model = JobModel(
             created_at=utils.utcnow(),
             name=name,
@@ -247,7 +245,7 @@ def create(
             stopped_callback_name=on_stopped.name if on_stopped else None,
             stopped_callback_timeout=on_stopped.timeout if on_stopped else None,
             success_ttl=result_ttl,
-            job_info_ttl=job_info_ttl or SCHEDULER_CONFIG.DEFAULT_JOB_TTL,
+            job_info_ttl=job_info_ttl,
             timeout=timeout,
             status=status,
             last_heartbeat=None,
diff --git a/scheduler/redis_models/registry/queue_registries.py b/scheduler/redis_models/registry/queue_registries.py
@@ -81,8 +81,7 @@ def get_jobs_to_schedule(self, timestamp: int, chunk_size: int = 1000) -> List[s
         :param chunk_size: Max results to return.
         :returns: A list of job names
         """
-        score = timestamp
-        jobs_to_schedule = self.connection.zrangebyscore(self._key, 0, score, start=0, num=chunk_size)
+        jobs_to_schedule = self.connection.zrangebyscore(self._key, 0, max=timestamp, start=0, num=chunk_size)
         return [as_str(job_name) for job_name in jobs_to_schedule]
 
     def get_scheduled_time(self, job_name: str) -> datetime:
diff --git a/scheduler/settings_types.py b/scheduler/settings_types.py
@@ -66,6 +66,14 @@ class QueueConfiguration:
     MASTER_NAME: Optional[str] = None
     CONNECTION_KWARGS: Optional[Dict[str, Any]] = None
 
+    def __post_init__(self):
+        if not any((self.URL, self.UNIX_SOCKET_PATH, self.HOST, self.SENTINELS)):
+            raise ValueError(f"At least one of URL, UNIX_SOCKET_PATH, HOST must be provided: {self}")
+        if sum((self.URL is not None, self.UNIX_SOCKET_PATH is not None, self.HOST is not None)) > 1:
+            raise ValueError(f"Only one of URL, UNIX_SOCKET_PATH, HOST should be provided: {self}")
+        if self.HOST is not None and (self.PORT is None or self.DB is None):
+            raise ValueError(f"HOST requires PORT and DB: {self}")
+
     def same_connection_params(self, other: Self) -> bool:
         for field in self.__CONNECTION_FIELDS__:
             if getattr(self, field) != getattr(other, field):
diff --git a/scheduler/tests/jobs.py b/scheduler/tests/jobs.py
@@ -35,7 +35,6 @@ def failing_job():
 
 
 def test_job():
-    # sleep(3)
     return 1 + 1
 
 
diff --git a/scheduler/tests/test_internals.py b/scheduler/tests/test_internals.py
@@ -24,6 +24,12 @@ def test_callback_bad_arguments(self):
         with self.assertRaises(CallbackSetupError) as cm:
             Callback("scheduler.tests.jobs.non_existing_method")
         self.assertEqual(str(cm.exception), "Invalid attribute name: 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")
+        with self.assertRaises(CallbackSetupError) as cm:
+            Callback("non_existing_method")
+        self.assertEqual(str(cm.exception), "Invalid attribute name: 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")
diff --git a/scheduler/tests/test_views/test_queue_actions.py b/scheduler/tests/test_views/test_queue_actions.py
@@ -15,7 +15,7 @@ def test_job_list_action_delete_jobs__with_bad_next_url(self):
         # enqueue some jobs
         job_names = []
         for _ in range(0, 3):
-            job = queue.create_and_enqueue_job(test_job)
+            job = queue.create_and_enqueue_job(test_job, job_info_ttl=0)
             job_names.append(job.name)
 
         # remove those jobs using view
@@ -46,7 +46,7 @@ def test_job_list_action_delete_jobs(self):
         # enqueue some jobs
         job_names = []
         for _ in range(0, 3):
-            job = queue.create_and_enqueue_job(test_job)
+            job = queue.create_and_enqueue_job(test_job, job_info_ttl=0)
             job_names.append(job.name)
 
         # remove those jobs using view
diff --git a/scheduler/tests/test_views/test_queue_job_action.py b/scheduler/tests/test_views/test_queue_job_action.py
@@ -37,8 +37,8 @@ def test_single_job_action_requeue_job(self):
         queue.delete_job(job.name)
 
     def test_single_job_action_delete_job(self):
-        queue = get_queue("django_tasks_scheduler_test")
-        job = queue.create_and_enqueue_job(test_job)
+        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)
         self.assertEqual(200, res.status_code)
         self.client.post(reverse("queue_job_action", args=[job.name, "delete"]), {"post": "yes"}, follow=True)
@@ -92,3 +92,29 @@ def test_single_job_action_enqueue_job(self):
         tmp = JobModel.get(job_list[-1].name, connection=queue.connection)
         self.assertEqual(tmp.get_status(connection=queue.connection), JobStatus.QUEUED)
         self.assertIsNotNone(tmp.enqueued_at)
+
+
+
+    def test_single_job_action_enqueue_job_sync_queue(self):
+        queue = get_queue("scheduler_scheduler_active_test")
+        job_list = []
+        # enqueue some jobs that depends on other
+        for _ in range(0, 3):
+            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.FAILED)
+        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)
+        self.assertEqual(200, res.status_code)
+        res = self.client.post(reverse("queue_job_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)
+        self.assertEqual(tmp.get_status(connection=queue.connection), JobStatus.FINISHED)
+        self.assertIsNotNone(tmp.enqueued_at)
+
diff --git a/scheduler/tests/test_worker/test_scheduler.py b/scheduler/tests/test_worker/test_scheduler.py
@@ -1,13 +1,30 @@
+import time
+from datetime import timedelta, datetime
+
 from scheduler.helpers.tools import create_worker
-from scheduler.tests.testtools import SchedulerBaseCase
+from scheduler.models.task import TaskType
+from scheduler.tests.testtools import SchedulerBaseCase, task_factory
 
 
-class TestWorker(SchedulerBaseCase):
+class TestWorkerScheduler(SchedulerBaseCase):
     def test_create_worker_with_scheduler__scheduler_started(self):
         worker = create_worker("default", name="test", burst=True, with_scheduler=True)
-        self.assertEqual(worker.name, "test")
         worker.bootstrap()
         self.assertIsNotNone(worker.scheduler)
         worker.stop_scheduler()
+        self.assertIsNone(worker.scheduler)
 
-
+    def test_scheduler_schedules_tasks(self):
+        task = task_factory(TaskType.ONCE, scheduled_time=datetime.now() + timedelta(milliseconds=40))
+        self.assertIsNotNone(task.job_id)
+        self.assertNotIn(task.job_id, task.rqueue.queued_job_registry)
+        self.assertIn(task.job_id, task.rqueue.scheduled_job_registry)
+        worker = create_worker("default", name="test", burst=True, with_scheduler=True)
+        worker.bootstrap()
+        time.sleep(1)
+        worker.scheduler.enqueue_scheduled_jobs()
+        self.assertIsNotNone(task.job_id)
+        self.assertNotIn(task.job_id, task.rqueue.scheduled_job_registry)
+        self.assertIn(task.job_id, task.rqueue.queued_job_registry)
+        worker.stop_scheduler()
+        self.assertIsNone(worker.scheduler)
diff --git a/scheduler/tests/testtools.py b/scheduler/tests/testtools.py
@@ -138,7 +138,3 @@ def tearDown(self) -> None:
         super(SchedulerBaseCase, self).tearDown()
         queue = get_queue("default")
         queue.empty()
-
-    @classmethod
-    def setUpClass(cls):
-        super(SchedulerBaseCase, cls).setUpClass()
diff --git a/scheduler/views/helpers.py b/scheduler/views/helpers.py
@@ -30,7 +30,6 @@ def _find_job(job_name: str) -> Tuple[Optional[Queue], Optional[JobModel]]:
                 return queue, job
         except Exception as e:
             _QUEUES_WITH_BAD_CONFIGURATION.add(queue_name)
-            logger.debug(f"Got exception: {e}")
-            logger.debug(f"Queue {queue_name} added to bad configuration")
+            logger.debug(f"Queue {queue_name} added to bad configuration - Got exception: {e}")
             pass
     return None, None
diff --git a/scheduler/views/job_views.py b/scheduler/views/job_views.py
@@ -72,12 +72,11 @@ def job_action(request: HttpRequest, job_name: str, action: str) -> HttpResponse
                 messages.info(request, f"You have successfully re-queued {job.name}")
             return redirect("job_details", job_name)
         elif action == "delete":
-            # Remove job name from queue and delete the actual job
             queue.delete_job(job.name)
             messages.info(request, f"You have successfully deleted {job.name}")
             return redirect("queue_registry_jobs", queue.name, "queued")
         elif action == "enqueue":
-            queue.delete_job(job.name)
+            queue.delete_job(job.name, expire_job_model=False)
             queue.enqueue_job(job)
             messages.info(request, f"You have successfully enqueued {job.name}")
             return redirect("job_details", job_name)
diff --git a/scheduler/views/queue_views.py b/scheduler/views/queue_views.py
@@ -199,7 +199,7 @@ def queue_confirm_action(request: HttpRequest, queue_name: str) -> HttpResponse:
     }
     return render(request, "admin/scheduler/confirm_action.html", context_data)
 
-
+_QUEUE_ACTIONS = {"delete", "requeue", "stop"}
 @never_cache
 @staff_member_required
 def queue_actions(request: HttpRequest, queue_name: str) -> HttpResponse:
@@ -215,7 +215,6 @@ def queue_actions(request: HttpRequest, queue_name: str) -> HttpResponse:
         for job in jobs:
             if job is None:
                 continue
-            # Remove job name from queue and delete the actual job
             queue.delete_job(job.name)
         messages.info(request, f"You have successfully deleted {len(job_names)} jobs!")
     elif action == "requeue":
@@ -230,7 +229,6 @@ def queue_actions(request: HttpRequest, queue_name: str) -> HttpResponse:
             try:
                 command = StopJobCommand(job_name=job.name, worker_name=job.worker_name)
                 send_command(connection=queue.connection, command=command)
-
                 queue.cancel_job(job.name)
                 cancelled_jobs += 1
             except Exception as e:
diff --git a/scheduler/worker/scheduler.py b/scheduler/worker/scheduler.py
@@ -77,14 +77,15 @@ def _acquire_locks(self) -> Set[str]:
             lock = SchedulerLock(queue_name)
             if lock.acquire(self.pid, connection=self.connection, expire=self.interval + 60):
                 self._locks[queue_name] = lock
+                successful_locks.add(queue_name)
 
         # Always reset _scheduled_job_registries when acquiring locks
         self.lock_acquisition_time = datetime.now()
         self._scheduled_job_registries = []
         for queue_name in self._locks:
             queue = get_queue(queue_name)
             self._scheduled_job_registries.append(queue.scheduled_job_registry)
-
+        logger.debug(f"[Scheduler {self.worker_name}/{self.pid}] Locks acquired for {', '.join(self._locks.keys())}")
         return successful_locks
 
     def start(self, burst=False) -> None:
@@ -159,12 +160,11 @@ def enqueue_scheduled_jobs(self) -> None:
 
             queue = get_queue(registry.name)
 
+            jobs = JobModel.get_many(job_names, connection=self.connection)
             with self.connection.pipeline() as pipeline:
-                jobs = JobModel.get_many(job_names, connection=self.connection)
                 for job in jobs:
                     if job is not None:
                         queue.enqueue_job(job, connection=pipeline, at_front=bool(job.at_front))
-                        registry.delete(pipeline, job.name)
                 pipeline.execute()
         self._status = SchedulerStatus.STARTED
 
diff --git a/scheduler/worker/worker.py b/scheduler/worker/worker.py

Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,6 @@ def failing_job():`
`35`	`35`
`36`	`36`
`37`	`37`	`def test_job():`
`38`		`- # sleep(3)`
`39`	`38`	`return 1 + 1`
`40`	`39`
`41`	`40`