test:workers cleanup

Author: cunla

scheduler/helpers/queues/queue_logic.py

@@ -16,7 +16,6 @@
     CanceledJobRegistry,
     ScheduledJobRegistry,
     QueuedJobRegistry,
-    NoSuchJobError,
 )
 from scheduler.redis_models import JobStatus, SchedulerLock, Result, ResultType, JobModel
 from scheduler.settings import logger, SCHEDULER_CONFIG
@@ -27,6 +26,10 @@ class InvalidJobOperation(Exception):
     pass
 
 
+class NoSuchJobError(Exception):
+    pass
+
+
 def perform_job(job_model: JobModel, connection: ConnectionType) -> Any:  # noqa
     """The main execution method. Invokes the job function with the job arguments.
 

scheduler/redis_models/__init__.py

@@ -15,7 +15,6 @@
     "CanceledJobRegistry",
     "ScheduledJobRegistry",
     "QueuedJobRegistry",
-    "NoSuchJobError",
 ]
 
 from .base import as_str
@@ -29,7 +28,6 @@
     CanceledJobRegistry,
     ScheduledJobRegistry,
     QueuedJobRegistry,
-    NoSuchJobError,
 )
 from .result import Result, ResultType
 from .worker import WorkerModel

scheduler/redis_models/registry/queue_registries.py

@@ -9,10 +9,6 @@
 from ..job import JobModel
 
 
-class NoSuchJobError(Exception):
-    pass
-
-
 class QueuedJobRegistry(JobNamesRegistry):
     _element_key_template: ClassVar[str] = ":registry:{}:queued_jobs"
 
@@ -87,17 +83,16 @@ def get_jobs_to_schedule(self, timestamp: int, chunk_size: int = 1000) -> List[s
         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:
+    def get_scheduled_time(self, job_name: str) -> Optional[datetime]:
         """Returns datetime (UTC) at which job is scheduled to be enqueued
 
         :param job_name: Job name
-        :raises NoSuchJobError: If the job was not found
-        :returns: The scheduled time as datetime object
+        :returns: The scheduled time as datetime object, or None if job is not found
         """
 
-        score = self.connection.zscore(self._key, job_name)
+        score: Optional[float] = self.connection.zscore(self._key, job_name)
         if not score:
-            raise NoSuchJobError
+            return None
 
         return datetime.fromtimestamp(score, tz=timezone.utc)
 

scheduler/templatetags/scheduler_tags.py

@@ -68,7 +68,4 @@ def job_runtime(job: JobModel):
 
 @register.filter
 def job_scheduled_time(job: JobModel, queue: Queue):
-    try:
-        return queue.scheduled_job_registry.get_scheduled_time(job.name)
-    except Exception:
-        return None
+    return queue.scheduled_job_registry.get_scheduled_time(job.name)