chore:ruff formating

Author: cunla

scheduler/__init__.py

@@ -9,4 +9,4 @@
 ]
 
 from scheduler.settings_types import QueueConfiguration, SchedulerConfig
-from scheduler.decorators import job
+from scheduler.decorators import job

scheduler/admin/task_admin.py

@@ -22,9 +22,9 @@ def job_execution_of(job: JobModel, task: Task) -> bool:
 def get_job_executions_for_task(queue_name: str, scheduled_task: Task) -> List[JobModel]:
     queue = get_queue(queue_name)
     job_list: List[JobModel] = JobModel.get_many(queue.get_all_job_names(), connection=queue.connection)
-    res = sorted(list(filter(lambda j: job_execution_of(j, scheduled_task), job_list)),
-                 key=lambda j: j.created_at,
-                 reverse=True)
+    res = sorted(
+        list(filter(lambda j: job_execution_of(j, scheduled_task), job_list)), key=lambda j: j.created_at, reverse=True
+    )
     return res
 
 
@@ -50,7 +50,10 @@ class TaskAdmin(admin.ModelAdmin):
     """TaskAdmin admin view for all task models."""
 
     class Media:
-        js = ("admin/js/jquery.init.js", "admin/js/select-fields.js",)
+        js = (
+            "admin/js/jquery.init.js",
+            "admin/js/select-fields.js",
+        )
 
     save_on_top = True
     change_form_template = "admin/scheduler/change_form.html"
@@ -90,18 +93,41 @@ class Media:
     fieldsets = (
         (
             None,
-            dict(fields=(
-                "name", "callable",
-                ("enabled", "timeout", "result_ttl"),
-                "task_type",
-            )),
+            dict(
+                fields=(
+                    "name",
+                    "callable",
+                    ("enabled", "timeout", "result_ttl"),
+                    "task_type",
+                )
+            ),
+        ),
+        (
+            None,
+            dict(fields=("scheduled_time",), classes=("tasktype-OnceTaskType",)),
+        ),
+        (
+            None,
+            dict(fields=("cron_string",), classes=("tasktype-CronTaskType",)),
+        ),
+        (
+            None,
+            dict(
+                fields=(
+                    (
+                        "interval",
+                        "interval_unit",
+                    ),
+                    "repeat",
+                ),
+                classes=("tasktype-RepeatableTaskType",),
+            ),
         ),
-        (None, dict(fields=("scheduled_time",), classes=("tasktype-OnceTaskType",)),),
-        (None, dict(fields=("cron_string",), classes=("tasktype-CronTaskType",)),),
-        (None, dict(fields=(("interval", "interval_unit",), "repeat"), classes=("tasktype-RepeatableTaskType",)),),
         (_("Queue settings"), dict(fields=(("queue", "at_front"), "job_name"))),
-        (_("Previous runs info"),
-         dict(fields=(("successful_runs", "last_successful_run"), ("failed_runs", "last_failed_run"))),),
+        (
+            _("Previous runs info"),
+            dict(fields=(("successful_runs", "last_successful_run"), ("failed_runs", "last_failed_run"))),
+        ),
     )
 
     @admin.display(description="Schedule")

scheduler/decorators.py

@@ -9,18 +9,18 @@
 
 class job:
     def __init__(
-            self,
-            queue: Union["Queue", str, None] = None,  # noqa: F821
-            connection: Optional[ConnectionType] = None,
-            timeout: Optional[int] = None,
-            result_ttl: Optional[int] = None,
-            job_info_ttl: Optional[int] = None,
-            at_front: bool = False,
-            meta: Optional[Dict[Any, Any]] = None,
-            description: Optional[str] = None,
-            on_failure: Optional[Union["Callback", Callable[..., Any]]] = None,
-            on_success: Optional[Union["Callback", Callable[..., Any]]] = None,
-            on_stopped: Optional[Union["Callback", Callable[..., Any]]] = None,
+        self,
+        queue: Union["Queue", str, None] = None,  # noqa: F821
+        connection: Optional[ConnectionType] = None,
+        timeout: Optional[int] = None,
+        result_ttl: Optional[int] = None,
+        job_info_ttl: Optional[int] = None,
+        at_front: bool = False,
+        meta: Optional[Dict[Any, Any]] = None,
+        description: Optional[str] = None,
+        on_failure: Optional[Union["Callback", Callable[..., Any]]] = None,
+        on_success: Optional[Union["Callback", Callable[..., Any]]] = None,
+        on_stopped: Optional[Union["Callback", Callable[..., Any]]] = None,
     ):
         """A decorator that adds a ``delay`` method to the decorated function, which in turn creates a RQ job when
         called. Accepts a required ``queue`` argument that can be either a ``Queue`` instance or a string
@@ -48,6 +48,7 @@ def __init__(
         :param on_stopped: Callable to run when stopped
         """
         from scheduler.helpers.queues import get_queue
+
         if queue is None:
             queue = "default"
         self.queue = get_queue(queue) if isinstance(queue, str) else queue

scheduler/helpers/callback.py

@@ -55,7 +55,7 @@ def _import_attribute(name: str) -> Callable[..., Any]:
         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:

scheduler/helpers/queues/queue_logic.py

@@ -117,7 +117,6 @@ def clean_registries(self, timestamp: Optional[float] = None) -> None:
                     logger.exception(f"Job {self.name}: error while executing failure callback")
                     raise
 
-
                 else:
                     logger.warning(
                         f"{self.__class__.__name__} cleanup: Moving job to {self.failed_job_registry.key} "
@@ -189,24 +188,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)
@@ -258,23 +257,37 @@ def create_and_enqueue_job(
     def job_handle_success(self, job: JobModel, result: Any, result_ttl: int, connection: ConnectionType):
         """Saves and cleanup job after successful execution"""
         job.after_execution(
-            result_ttl, JobStatus.FINISHED,
+            result_ttl,
+            JobStatus.FINISHED,
             prev_registry=self.active_job_registry,
-            new_registry=self.finished_job_registry, connection=connection)
-        Result.create(connection, job_name=job.name, worker_name=job.worker_name, _type=ResultType.SUCCESSFUL,
-                      return_value=result, ttl=result_ttl)
+            new_registry=self.finished_job_registry,
+            connection=connection,
+        )
+        Result.create(
+            connection,
+            job_name=job.name,
+            worker_name=job.worker_name,
+            _type=ResultType.SUCCESSFUL,
+            return_value=result,
+            ttl=result_ttl,
+        )
 
     def job_handle_failure(self, status: JobStatus, job: JobModel, exc_string: str, connection: ConnectionType):
         # Does not set job status since the job might be stopped
         job.after_execution(
-            SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL, status,
+            SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL,
+            status,
             prev_registry=self.active_job_registry,
             new_registry=self.failed_job_registry,
-            connection=connection)
+            connection=connection,
+        )
         Result.create(
-            connection, job.name, job.worker_name,
-            ResultType.FAILED, SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL,
-            exc_string=exc_string
+            connection,
+            job.name,
+            job.worker_name,
+            ResultType.FAILED,
+            SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL,
+            exc_string=exc_string,
         )
 
     def run_job(self, job: JobModel) -> JobModel:
@@ -299,7 +312,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.
 
@@ -350,10 +363,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.
@@ -439,8 +452,9 @@ def cancel_job(self, job_name: str) -> None:
                 if new_status == JobStatus.CANCELED:
                     self.canceled_job_registry.add(pipe, job_name, 0)
                 else:
-                    self.finished_job_registry.add(pipe, job_name,
-                                                   current_timestamp() + SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL)
+                    self.finished_job_registry.add(
+                        pipe, job_name, current_timestamp() + SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL
+                    )
                 pipe.execute()
                 break
             except WatchError:

scheduler/models/task.py

@@ -189,9 +189,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/job.py

@@ -128,9 +128,14 @@ def prepare_for_execution(self, worker_name: str, registry: JobNamesRegistry, co
         registry.add(connection, self.name, self.last_heartbeat.timestamp())
         self.save(connection=connection)
 
-    def after_execution(self, result_ttl: int, status: JobStatus, connection: ConnectionType,
-                        prev_registry: Optional[JobNamesRegistry] = None,
-                        new_registry: Optional[JobNamesRegistry] = None) -> None:
+    def after_execution(
+        self,
+        result_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
         self.ended_at = utils.utcnow()
@@ -167,26 +172,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.
@@ -260,7 +265,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/result.py

@@ -31,19 +31,23 @@ 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,
-            ttl=ttl, type=_type, return_value=return_value, exc_string=exc_string,
-            worker_name=worker_name, )
+            ttl=ttl,
+            type=_type,
+            return_value=return_value,
+            exc_string=exc_string,
+            worker_name=worker_name,
+        )
         result.save(connection)
         return result
 

scheduler/tests/jobs.py

@@ -18,10 +18,12 @@ def test_args_kwargs(*args, **kwargs):
     kwargs_list = [f"{k}={v}" for (k, v) in kwargs.items()]
     return func.format(", ".join(args_list + kwargs_list))
 
+
 def two_seconds_job():
     sleep(2)
     logging.info(f"Job {_counter}")
 
+
 def long_job():
     sleep(1000)
     logging.info(f"Job {_counter}")