format

Author: cunla

.github/workflows/test.yml

@@ -30,7 +30,7 @@ jobs:
   test-regular:
     needs: [ 'ruff' ]
     runs-on: ubuntu-latest
-    name: "Run tests ${{ matrix.python-version }}/${{ matrix.django-version }}/${{ matrix.broker }}"
+    name: "Tests py${{ matrix.python-version }}/dj${{ matrix.django-version }}/${{ matrix.broker }}"
     strategy:
       max-parallel: 6
       matrix:

scheduler/helpers/queues/queue_logic.py

@@ -176,24 +176,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)
@@ -245,7 +245,7 @@ def create_and_enqueue_job(
         return job_model
 
     def job_handle_success(
-            self, job: JobModel, result: Any, job_info_ttl: int, result_ttl: int, connection: ConnectionType
+        self, job: JobModel, result: Any, job_info_ttl: int, result_ttl: int, connection: ConnectionType
     ):
         """Saves and cleanup job after successful execution"""
         job.after_execution(
@@ -304,10 +304,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.
@@ -410,7 +410,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/job.py

@@ -125,12 +125,12 @@ def prepare_for_execution(self, worker_name: str, registry: JobNamesRegistry, co
         self.save(connection=connection)
 
     def after_execution(
-            self,
-            job_info_ttl: int,
-            status: JobStatus,
-            connection: ConnectionType,
-            prev_registry: Optional[JobNamesRegistry] = None,
-            new_registry: Optional[JobNamesRegistry] = None,
+        self,
+        job_info_ttl: int,
+        status: JobStatus,
+        connection: ConnectionType,
+        prev_registry: Optional[JobNamesRegistry] = None,
+        new_registry: Optional[JobNamesRegistry] = None,
     ) -> None:
         """After the job is executed, update the status, heartbeat, and other metadata."""
         self.status = status
@@ -168,26 +168,26 @@ def get_call_string(self):
 
     @classmethod
     def create(
-            cls,
-            connection: ConnectionType,
-            func: FunctionReferenceType,
-            queue_name: str,
-            args: Union[List[Any], Optional[Tuple]] = None,
-            kwargs: Optional[Dict[str, Any]] = None,
-            result_ttl: Optional[int] = None,
-            job_info_ttl: Optional[int] = None,
-            status: Optional[JobStatus] = None,
-            description: Optional[str] = None,
-            timeout: Optional[int] = None,
-            name: Optional[str] = None,
-            task_type: Optional[str] = None,
-            scheduled_task_id: Optional[int] = None,
-            meta: Optional[Dict[str, Any]] = None,
-            *,
-            on_success: Optional[Callback] = None,
-            on_failure: Optional[Callback] = None,
-            on_stopped: Optional[Callback] = None,
-            at_front: Optional[bool] = None,
+        cls,
+        connection: ConnectionType,
+        func: FunctionReferenceType,
+        queue_name: str,
+        args: Union[List[Any], Optional[Tuple]] = None,
+        kwargs: Optional[Dict[str, Any]] = None,
+        result_ttl: Optional[int] = None,
+        job_info_ttl: Optional[int] = None,
+        status: Optional[JobStatus] = None,
+        description: Optional[str] = None,
+        timeout: Optional[int] = None,
+        name: Optional[str] = None,
+        task_type: Optional[str] = None,
+        scheduled_task_id: Optional[int] = None,
+        meta: Optional[Dict[str, Any]] = None,
+        *,
+        on_success: Optional[Callback] = None,
+        on_failure: Optional[Callback] = None,
+        on_stopped: Optional[Callback] = None,
+        at_front: Optional[bool] = None,
     ) -> Self:
         """Creates a new job-model for the given function, arguments, and keyword arguments.
         :returns: A job-model instance.
@@ -261,7 +261,7 @@ def create(
 
 
 def _get_call_string(
-        func_name: Optional[str], args: Any, kwargs: Dict[Any, Any], max_length: Optional[int] = None
+    func_name: Optional[str], args: Any, kwargs: Dict[Any, Any], max_length: Optional[int] = None
 ) -> Optional[str]:
     """
     Returns a string representation of the call, formatted as a regular

scheduler/redis_models/registry/base_registry.py

@@ -86,7 +86,7 @@ def key(self) -> str:
 
     @classmethod
     def pop(
-            cls, connection: ConnectionType, registries: Sequence[Self], timeout: Optional[int]
+        cls, connection: ConnectionType, registries: Sequence[Self], timeout: Optional[int]
     ) -> Tuple[Optional[str], Optional[str]]:
         """Helper method to abstract away from some Redis API details
 

scheduler/redis_models/result.py

@@ -31,14 +31,14 @@ class Result(StreamModel):
 
     @classmethod
     def create(
-            cls,
-            connection: ConnectionType,
-            job_name: str,
-            worker_name: str,
-            _type: ResultType,
-            ttl: int,
-            return_value: Any = None,
-            exc_string: Optional[str] = None,
+        cls,
+        connection: ConnectionType,
+        job_name: str,
+        worker_name: str,
+        _type: ResultType,
+        ttl: int,
+        return_value: Any = None,
+        exc_string: Optional[str] = None,
     ) -> Self:
         result = cls(
             parent=job_name,

scheduler/redis_models/worker.py

@@ -118,4 +118,4 @@ def _split_list(a_list: List[str], segment_size: int) -> Generator[list[str], An
     :returns: The list split into smaller lists
     """
     for i in range(0, len(a_list), segment_size):
-        yield a_list[i: i + segment_size]
+        yield a_list[i : i + segment_size]

scheduler/views/job_views.py

@@ -20,6 +20,7 @@ class JobDetailAction(str, Enum):
     ENQUEUE = "enqueue"
     CANCEL = "cancel"
 
+
 @never_cache
 @staff_member_required
 def job_detail(request: HttpRequest, job_name: str) -> HttpResponse: