ruff

Author: cunla

scheduler/admin/task_admin.py

@@ -15,10 +15,14 @@
 from scheduler.settings import SCHEDULER_CONFIG, logger
 
 
+def job_execution_of(job: JobModel, task: Task) -> bool:
+    return job.scheduled_task_id == task.id and job.task_type == task.task_type
+
+
 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.queued_job_registry.all(), connection=queue.connection)
-    res = sorted(list(filter(lambda j: j.is_execution_of(scheduled_task), job_list)),
+    res = sorted(list(filter(lambda j: job_execution_of(j, scheduled_task), job_list)),
                  key=lambda j: j.created_at,
                  reverse=True)
     return res

scheduler/decorators.py

@@ -1,15 +1,16 @@
 from functools import wraps
 from typing import Any, Callable, Dict, List, Optional, Union
 
-from scheduler.broker_types import ConnectionType, FunctionReferenceType
+from scheduler.broker_types import ConnectionType
+from scheduler.helpers.callback import Callback
 
 JOB_METHODS_LIST = list()
 
 
 class job:
     def __init__(
             self,
-            queue: Union["Queue", str, None] = None,
+            queue: Union["Queue", str, None] = None,  # noqa: F821
             connection: Optional[ConnectionType] = None,
             timeout: Optional[int] = None,
             result_ttl: Optional[int] = None,

scheduler/helpers/callback.py

@@ -2,7 +2,6 @@
 import inspect
 from typing import Union, Callable, Any, Optional
 
-from scheduler.settings import SCHEDULER_CONFIG
 from scheduler.timeouts import JobTimeoutException
 
 
@@ -12,6 +11,8 @@ class CallbackSetupError(Exception):
 
 class Callback:
     def __init__(self, func: Union[str, Callable[..., Any]], timeout: Optional[int] = None):
+        from scheduler.settings import SCHEDULER_CONFIG
+
         self.timeout = timeout or SCHEDULER_CONFIG.CALLBACK_TIMEOUT
         if not isinstance(self.timeout, int) or self.timeout < 0:
             raise CallbackSetupError(f"Callback `timeout` must be a positive int, but received {self.timeout}")
@@ -26,6 +27,8 @@ def name(self) -> str:
         return "{0}.{1}".format(self.func.__module__, self.func.__qualname__)
 
     def __call__(self, *args, **kwargs):
+        from scheduler.settings import SCHEDULER_CONFIG
+
         with SCHEDULER_CONFIG.DEATH_PENALTY_CLASS(self.timeout, JobTimeoutException):
             return self.func(*args, **kwargs)
 

scheduler/helpers/queues/queue_logic.py

@@ -259,7 +259,7 @@ def create_and_enqueue_job(
             scheduled_task_id=scheduled_task_id,
         )
         if when is None:
-            job_model = self._enqueue_job(job_model, connection=pipeline, at_front=at_front)
+            job_model = self.enqueue_job(job_model, connection=pipeline, at_front=at_front)
         else:
             job_model.save(connection=self.connection)
             self.scheduled_job_registry.schedule(self.connection, job_model, when)
@@ -287,7 +287,7 @@ def job_handle_failure(self, status: JobStatus, job: JobModel, exc_string: str,
             exc_string=exc_string
         )
 
-    def run_job(self, job: JobModel) -> Any:
+    def run_job(self, job: JobModel) -> JobModel:
         """Run the job
         :param job: The job to run
         :returns: The job result
@@ -300,22 +300,22 @@ def run_job(self, job: JobModel) -> Any:
                 self.job_handle_success(job, result=result, result_ttl=result_ttl, connection=pipeline)
                 job.expire(result_ttl, connection=pipeline)
                 pipeline.execute()
-            return result
-        except:
+        except Exception as e:
+            logger.warning(f"Job {job.name} failed with exception: {e}")
             with self.connection.pipeline() as pipeline:
                 exc_string = "".join(traceback.format_exception(*sys.exc_info()))
                 self.job_handle_failure(JobStatus.FAILED, job, exc_string, pipeline)
                 pipeline.execute()
+        return job
 
     def retry_job(self, job: JobModel, connection: ConnectionType):
         """Requeue or schedule this job for execution.
         If the the `retry_interval` was set on the job itself,
         it will calculate a scheduled time for the job to run, and instead
         of just regularly `enqueing` the job, it will `schedule` it.
 
-        Args:
-            job (JobModel): The queue to retry the job on
-            connection (ConnectionType): The Redis' pipeline to use
+        :param job: The job to retry
+        :param connection: The broker connection to use
         """
         number_of_intervals = len(job.retry_intervals)
         index = max(number_of_intervals - job.retries_left, 0)
@@ -327,9 +327,9 @@ def retry_job(self, job: JobModel, connection: ConnectionType):
             job.save(connection=connection)
             self.scheduled_job_registry.schedule(connection, job, scheduled_datetime)
         else:
-            self._enqueue_job(job, connection=connection)
+            self.enqueue_job(job, connection=connection)
 
-    def _enqueue_job(
+    def enqueue_job(
             self, job_model: JobModel, connection: Optional[ConnectionType] = None, at_front: bool = False
     ) -> JobModel:
         """Enqueues a job for delayed execution without checking dependencies.
@@ -369,7 +369,7 @@ def run_sync(self, job: JobModel) -> JobModel:
         job.prepare_for_execution("sync", self.active_job_registry, self.connection)
 
         try:
-            result = self.run_job(job)
+            self.run_job(job)
         except:  # noqa
             with self.connection.pipeline() as pipeline:
                 exc_string = "".join(traceback.format_exception(*sys.exc_info()))
@@ -503,7 +503,7 @@ def requeue_jobs(self, *job_names: str, at_front: bool = False) -> int:
                 job.started_at = None
                 job.ended_at = None
                 job.save(connection=pipe)
-                self._enqueue_job(job, connection=pipe, at_front=at_front)
+                self.enqueue_job(job, connection=pipe, at_front=at_front)
                 jobs_requeued += 1
             pipe.execute()
         return jobs_requeued

scheduler/management/commands/scheduler_worker.py

@@ -134,6 +134,7 @@ def handle(self, **options):
         # Verbosity is defined by default in BaseCommand for all commands
         verbosity = options.pop("verbosity", 3)
         log_level = VERBOSITY_TO_LOG_LEVEL.get(verbosity, logging.INFO)
+        logger.setLevel(log_level)
         logger.setLevel(logging.DEBUG)
         logging.basicConfig(level=logging.DEBUG)
 

scheduler/models/task.py

@@ -1,5 +1,4 @@
 import math
-import uuid
 from datetime import timedelta, datetime
 from typing import Dict, Any, Optional
 
@@ -190,9 +189,9 @@ def is_scheduled(self) -> bool:
             return False
         # check whether job_id is in scheduled/queued/active jobs
         res = (
-            (self.job_id in self.rqueue.scheduled_job_registry.all())
-            or (self.job_id in self.rqueue.queued_job_registry.all())
-            or (self.job_id in self.rqueue.active_job_registry.all())
+                (self.job_id in self.rqueue.scheduled_job_registry.all())
+                or (self.job_id in self.rqueue.queued_job_registry.all())
+                or (self.job_id 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

@@ -85,9 +85,6 @@ def set_status(self, status: JobStatus, connection: ConnectionType) -> None:
         """Set's the Job Status"""
         self.set_field("status", status, connection=connection)
 
-    def is_execution_of(self, task: "Task") -> bool:
-        return self.scheduled_task_id == task.id and self.task_type == task.task_type
-
     @property
     def is_queued(self) -> bool:
         return self.status == JobStatus.QUEUED
@@ -168,7 +165,7 @@ def stopped_callback(self) -> Optional[Callable[..., Any]]:
         return Callback(self.stopped_callback_name, self.stopped_callback_timeout)
 
     def get_call_string(self):
-        return get_call_string(self.func_name, self.args, self.kwargs)
+        return _get_call_string(self.func_name, self.args, self.kwargs)
 
     @classmethod
     def create(
@@ -233,7 +230,7 @@ def create(
             _func_name = "__call__"
         else:
             raise TypeError("Expected a callable or a string, but got: {0}".format(func))
-        description = description or get_call_string(func, args or [], kwargs or {}, max_length=75)
+        description = description or _get_call_string(func, args or [], kwargs or {}, max_length=75)
 
         if retries_left is not None and retries_left < 1:
             raise ValueError("max: please enter a value greater than 0")
@@ -283,22 +280,19 @@ def create(
         return model
 
 
-def get_call_string(
+def _get_call_string(
         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
     Python function invocation statement. If max_length is not None, truncate
     arguments with representation longer than max_length.
 
-    Args:
-        func_name (str): The funtion name
-        args (Any): The function arguments
-        kwargs (Dict[Any, Any]): The function kwargs
-        max_length (int, optional): The max length. Defaults to None.
-
-    Returns:
-        str: A String representation of the function call.
+    :param func_name: The function name
+    :param args: The function arguments
+    :param kwargs: The function kwargs
+    :param max_length: The max length of the return string
+    :return: A string representation of the function call
     """
     if func_name is None:
         return None

scheduler/tests/test_views/test_queue_job_action.py

@@ -74,11 +74,9 @@ def test_single_job_action_enqueue_job(self):
         queue = get_queue("django_tasks_scheduler_test")
         job_list = []
         # enqueue some jobs that depends on other
-        previous_job = None
         for _ in range(0, 3):
             job = queue.create_and_enqueue_job(test_job)
             job_list.append(job)
-            previous_job = job
 
         # This job is deferred
 

scheduler/tests/testtools.py

@@ -9,6 +9,7 @@
 from django.utils import timezone
 
 from scheduler import settings
+from scheduler.admin.task_admin import job_execution_of
 from scheduler.helpers.queues import get_queue
 from scheduler.helpers.tools import create_worker
 from scheduler.models.args import TaskKwarg
@@ -113,10 +114,10 @@ def _get_task_scheduled_job_from_registry(django_task: Task) -> JobModel:
     return JobModel.get(entry, connection=django_task.rqueue.connection)
 
 
-def _get_executions(django_job: Task):
-    job_ids = django_job.rqueue.get_all_job_names()
-    job_list: List[JobModel] = JobModel.get_many(job_ids, connection=django_job.rqueue.connection)
-    return list(filter(lambda j: j.is_execution_of(django_job), job_list))
+def _get_executions(task: Task):
+    job_ids = task.rqueue.get_all_job_names()
+    job_list: List[JobModel] = JobModel.get_many(job_ids, connection=task.rqueue.connection)
+    return list(filter(lambda j: job_execution_of(j, task), job_list))
 
 
 class SchedulerBaseCase(TestCase):

scheduler/views/job_views.py

@@ -78,7 +78,7 @@ def job_action(request: HttpRequest, job_name: str, action: str) -> HttpResponse
             return redirect("queue_registry_jobs", queue.name, "queued")
         elif action == "enqueue":
             queue.delete_job(job.name)
-            queue._enqueue_job(job)
+            queue.enqueue_job(job)
             messages.info(request, f"You have successfully enqueued {job.name}")
             return redirect("job_details", job_name)
         elif action == "cancel":