ruff

Author: cunla

scheduler/helpers/queues/queue_logic.py

@@ -186,24 +186,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,
-            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,
-            when: Optional[datetime] = None,
-            pipeline: Optional[ConnectionType] = None,
+        self,
+        func: FunctionReferenceType,
+        args: Union[Tuple, List, None] = None,
+        kwargs: Optional[Dict] = 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,
+        when: Optional[datetime] = 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)
@@ -310,7 +310,7 @@ def run_job(self, job: JobModel) -> JobModel:
         return job
 
     def enqueue_job(
-            self, job_model: JobModel, connection: Optional[ConnectionType] = None, at_front: bool = False
+        self, job_model: JobModel, connection: Optional[ConnectionType] = None, at_front: bool = False
     ) -> JobModel:
         """Enqueues a job for delayed execution without checking dependencies.
 
@@ -361,10 +361,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.

scheduler/models/task.py

@@ -186,9 +186,9 @@ def is_scheduled(self) -> bool:
             return False
         # check whether job_id is in scheduled/queued/active jobs
         res = (
-                (self.job_name in self.rqueue.scheduled_job_registry.all())
-                or (self.job_name in self.rqueue.queued_job_registry.all())
-                or (self.job_name in self.rqueue.active_job_registry.all())
+            (self.job_name in self.rqueue.scheduled_job_registry.all())
+            or (self.job_name in self.rqueue.queued_job_registry.all())
+            or (self.job_name in self.rqueue.active_job_registry.all())
         )
         # If the job_id is not scheduled/queued/started,
         # update the job_id to None. (The job_id belongs to a previous run which is completed)

scheduler/redis_models/registry/base_registry.py

@@ -30,6 +30,7 @@ def add(self, connection: ConnectionType, job_name: str, score: float, update_ex
     def delete(self, connection: ConnectionType, job_name: str) -> None:
         connection.zrem(self._key, job_name)
 
+
 class JobNamesRegistry(ZSetModel):
     _element_key_template: ClassVar[str] = ":registry:{}"
 

scheduler/tests/test_mgmt_commands/test_scheduler_worker.py

@@ -3,6 +3,7 @@
 
 from scheduler.helpers.queues import get_queue
 from scheduler.redis_models import JobModel
+from scheduler.settings import SCHEDULER_CONFIG
 from scheduler.tests import test_settings  # noqa
 from scheduler.tests.jobs import failing_job
 
@@ -26,6 +27,7 @@ def test_scheduler_worker__no_queues_params(self):
             self.assertTrue(job.is_failed)
 
     def test_scheduler_worker__run_jobs(self):
+        SCHEDULER_CONFIG.SCHEDULER_INTERVAL = 1
         queue = get_queue("default")
 
         # enqueue some jobs that will fail

scheduler/types/settings_types.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from enum import Enum
-from typing import Callable, Dict, Optional, List, Tuple, Any, Self, Type, Iterator
+from typing import Callable, Dict, Optional, List, Tuple, Any, Self, Type
 
 from scheduler.timeouts import BaseDeathPenalty, UnixSignalDeathPenalty
 
@@ -36,6 +36,7 @@ class SchedulerConfiguration:
     SCHEDULER_FALLBACK_PERIOD_SECS: int = 120  # Period (secs) to wait before requiring to reacquire locks
     DEATH_PENALTY_CLASS: Type["BaseDeathPenalty"] = UnixSignalDeathPenalty
 
+
 @dataclass(slots=True, frozen=True, kw_only=True)
 class QueueConfiguration:
     __CONNECTION_FIELDS__ = {

scheduler/worker/worker.py

@@ -20,7 +20,7 @@
 import scheduler
 from scheduler.helpers.queues import get_queue
 from scheduler.helpers.queues import get_queues
-from scheduler.redis_models import WorkerModel, JobModel, JobStatus, KvLock, DequeueTimeout
+from scheduler.redis_models import WorkerModel, JobModel, JobStatus, DequeueTimeout
 from scheduler.settings import SCHEDULER_CONFIG, logger
 from scheduler.types import Broker
 from scheduler.types import (
@@ -106,18 +106,18 @@ def from_model(cls, model: WorkerModel) -> Self:
         return res
 
     def __init__(
-            self,
-            queues,
-            name: str,
-            connection: Optional[ConnectionType] = None,
-            maintenance_interval: int = SCHEDULER_CONFIG.DEFAULT_MAINTENANCE_TASK_INTERVAL,
-            job_monitoring_interval=SCHEDULER_CONFIG.DEFAULT_JOB_MONITORING_INTERVAL,
-            dequeue_strategy: DequeueStrategy = DequeueStrategy.DEFAULT,
-            disable_default_exception_handler: bool = False,
-            fork_job_execution: bool = True,
-            with_scheduler: bool = True,
-            burst: bool = False,
-            model: Optional[WorkerModel] = None,
+        self,
+        queues,
+        name: str,
+        connection: Optional[ConnectionType] = None,
+        maintenance_interval: int = SCHEDULER_CONFIG.DEFAULT_MAINTENANCE_TASK_INTERVAL,
+        job_monitoring_interval=SCHEDULER_CONFIG.DEFAULT_JOB_MONITORING_INTERVAL,
+        dequeue_strategy: DequeueStrategy = DequeueStrategy.DEFAULT,
+        disable_default_exception_handler: bool = False,
+        fork_job_execution: bool = True,
+        with_scheduler: bool = True,
+        burst: bool = False,
+        model: Optional[WorkerModel] = None,
     ):  # noqa
         self.fork_job_execution = fork_job_execution
         self.job_monitoring_interval = job_monitoring_interval
@@ -209,9 +209,9 @@ def _install_signal_handlers(self) -> None:
         signal.signal(signal.SIGTERM, self.request_stop)
 
     def work(
-            self,
-            max_jobs: Optional[int] = None,
-            max_idle_time: Optional[int] = None,
+        self,
+        max_jobs: Optional[int] = None,
+        max_idle_time: Optional[int] = None,
     ) -> bool:
         """Starts the work loop.
 
@@ -383,7 +383,7 @@ def run_maintenance_tasks(self):
         self.clean_registries()
 
     def dequeue_job_and_maintain_ttl(
-            self, timeout: Optional[int], max_idle_time: Optional[int] = None
+        self, timeout: Optional[int], max_idle_time: Optional[int] = None
     ) -> Tuple[JobModel, Queue]:
         """Dequeues a job while maintaining the TTL.
         :param timeout: The timeout for the dequeue operation.
@@ -558,7 +558,7 @@ def reorder_queues(self, reference_queue: Queue):
             return
         if self._dequeue_strategy == DequeueStrategy.ROUND_ROBIN:
             pos = self._ordered_queues.index(reference_queue)
-            self._ordered_queues = self._ordered_queues[pos + 1:] + self._ordered_queues[: pos + 1]
+            self._ordered_queues = self._ordered_queues[pos + 1 :] + self._ordered_queues[: pos + 1]
             return
         if self._dequeue_strategy == DequeueStrategy.RANDOM:
             shuffle(self._ordered_queues)
@@ -642,7 +642,7 @@ def monitor_job_execution_process(self, job: JobModel, queue: Queue) -> None:
         while True:
             try:
                 with SCHEDULER_CONFIG.DEATH_PENALTY_CLASS(
-                        self.job_monitoring_interval, JobExecutionMonitorTimeoutException
+                    self.job_monitoring_interval, JobExecutionMonitorTimeoutException
                 ):
                     retpid, ret_val, rusage = self.wait_for_job_execution_process()
                 break
@@ -878,7 +878,7 @@ class RoundRobinWorker(Worker):
 
     def reorder_queues(self, reference_queue):
         pos = self._ordered_queues.index(reference_queue)
-        self._ordered_queues = self._ordered_queues[pos + 1:] + self._ordered_queues[: pos + 1]
+        self._ordered_queues = self._ordered_queues[pos + 1 :] + self._ordered_queues[: pos + 1]
 
 
 class RandomWorker(Worker):