refactor: separated queue views to several files

Author: cunla

scheduler/helpers/queues/queue_logic.py

@@ -173,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)
@@ -311,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.
@@ -417,7 +417,7 @@ def delete_job(self, job_name: str, expire_job_model: bool = True) -> None:
                 pass
 
     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.
 

scheduler/redis_models/registry/queue_registries.py

@@ -38,6 +38,7 @@ def empty(self) -> None:
                 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

@@ -32,7 +32,10 @@ def test_callback_bad_arguments(self):
         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), "Callback `func` is not callable: scheduler.tests.non_existing_module.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), "Callback `func` is not callable: non_existing_method")

scheduler/tests/testtools.py

@@ -47,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),

scheduler/views/helpers.py

@@ -46,9 +46,9 @@ def _check_next_url(request: HttpRequest, default_next_url: str) -> str:
     next_url = request.POST.get("next_url", default_next_url)
     next_url = next_url.replace("\\", "")
     if (
-            not url_has_allowed_host_and_scheme(next_url, allowed_hosts=None)
-            or urlparse(next_url).netloc
-            or urlparse(next_url).scheme
+        not url_has_allowed_host_and_scheme(next_url, allowed_hosts=None)
+        or urlparse(next_url).netloc
+        or urlparse(next_url).scheme
     ):
         messages.warning(request, "Bad followup URL")
         next_url = default_next_url

scheduler/views/queue_job_actions.py

@@ -1,4 +1,5 @@
 """list_registry_jobs actions on multiple selected jobs"""
+
 from enum import Enum
 
 from django.contrib import admin, messages

scheduler/views/queue_registry_actions.py

@@ -1,4 +1,5 @@
 """list_registry_jobs actions on all jobs in the registry"""
+
 from enum import Enum
 
 from django.contrib import admin, messages

scheduler/worker/scheduler.py

@@ -32,11 +32,11 @@ def _reschedule_tasks():
 
 class WorkerScheduler:
     def __init__(
-            self,
-            queues: Sequence[Queue],
-            connection: ConnectionType,
-            worker_name: str,
-            interval: Optional[int] = None,
+        self,
+        queues: Sequence[Queue],
+        connection: ConnectionType,
+        worker_name: str,
+        interval: Optional[int] = None,
     ) -> None:
         interval = interval or SCHEDULER_CONFIG.SCHEDULER_INTERVAL
         self._queues = queues