wip

Author: cunla

pyproject.toml

@@ -45,7 +45,7 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
-yaml = ["pyyaml~=6.0"]
+yaml = ["pyyaml~=6.0", "types-PyYAML>=6.0.12.20250516"]
 valkey = ["valkey>=6.0.2,<7"]
 sentry = ["sentry-sdk~=2.19"]
 
@@ -90,7 +90,10 @@ line-ending = "auto"
 
 [tool.mypy]
 packages = ['scheduler', ]
-exclude = "scheduler/tests/.*\\.py"
+exclude = ["scheduler/tests/.*\\.py",
+           "scheduler/migrations/.*\\.py",
+           "testproject/.*\\.py",
+           "testproject/tests/.*\\.py"]
 strict = true
 follow_imports = "silent"
 ignore_missing_imports = true

scheduler/helpers/sentry_integration.py

@@ -1,21 +1,23 @@
 import weakref
+from typing import Any, Callable
 
 import sentry_sdk
-from sentry_sdk.consts import OP
+from sentry_sdk._types import EventProcessor, Event, ExcInfo
 from sentry_sdk.api import continue_trace
-from sentry_sdk.integrations import _check_minimum_version, DidNotEnable, Integration
+from sentry_sdk.consts import OP
+from sentry_sdk.integrations import _check_minimum_version, Integration
 from sentry_sdk.integrations.logging import ignore_logger
 from sentry_sdk.tracing import TransactionSource
 from sentry_sdk.utils import (
     capture_internal_exceptions,
     ensure_integration_enabled,
     event_from_exception,
-    format_timestamp,
     parse_version,
 )
 
 import scheduler
 from scheduler.helpers.queues import Queue
+from scheduler.redis_models import JobModel
 from scheduler.redis_models import JobStatus
 from scheduler.timeouts import JobTimeoutException
 from scheduler.worker import Worker
@@ -26,36 +28,34 @@ class SentryIntegration(Integration):
     origin = f"auto.queue.{identifier}"
 
     @staticmethod
-    def setup_once():
-        # type: () -> None
+    def setup_once() -> None:
         version = parse_version(scheduler.__version__)
         _check_minimum_version(SentryIntegration, version)
 
         old_perform_job = Worker.perform_job
 
         @ensure_integration_enabled(SentryIntegration, old_perform_job)
-        def sentry_patched_perform_job(self, job, *args, **kwargs):
-            # type: (Any, Job, *Queue, **Any) -> bool
+        def sentry_patched_perform_job(self: Any, job_model: JobModel, *args: Queue, **kwargs: Any) -> bool:
             with sentry_sdk.new_scope() as scope:
                 scope.clear_breadcrumbs()
-                scope.add_event_processor(_make_event_processor(weakref.ref(job)))
+                scope.add_event_processor(_make_event_processor(weakref.ref(job_model)))
 
                 transaction = continue_trace(
-                    job.meta.get("_sentry_trace_headers") or {},
+                    job_model.meta.get("_sentry_trace_headers") or {},
                     op=OP.QUEUE_TASK_RQ,
                     name="unknown RQ task",
                     source=TransactionSource.TASK,
                     origin=SentryIntegration.origin,
                 )
 
                 with capture_internal_exceptions():
-                    transaction.name = job.func_name
+                    transaction.name = job_model.func_name
 
                 with sentry_sdk.start_transaction(
-                    transaction,
-                    custom_sampling_context={"rq_job": job},
+                        transaction,
+                        custom_sampling_context={"rq_job": job_model},
                 ):
-                    rv = old_perform_job(self, job, *args, **kwargs)
+                    rv = old_perform_job(self, job_model, *args, **kwargs)
 
             if self.is_horse:
                 # We're inside of a forked process and RQ is
@@ -65,30 +65,28 @@ def sentry_patched_perform_job(self, job, *args, **kwargs):
 
             return rv
 
-        Worker.perform_job = sentry_patched_perform_job
+        Worker.perform_job = sentry_patched_perform_job  # type: ignore[method-assign]
 
         old_handle_exception = Worker.handle_exception
 
-        def sentry_patched_handle_exception(self, job, *exc_info, **kwargs):
-            # type: (Worker, Any, *Any, **Any) -> Any
+        def sentry_patched_handle_exception(self: Worker, job: Any, *exc_info: Any, **kwargs: Any) -> Any:
             retry = (
-                hasattr(job, "retries_left")
-                and job.retries_left
-                and job.retries_left > 0
+                    hasattr(job, "retries_left")
+                    and job.retries_left
+                    and job.retries_left > 0
             )
             failed = job._status == JobStatus.FAILED or job.is_failed
             if failed and not retry:
                 _capture_exception(exc_info)
 
             return old_handle_exception(self, job, *exc_info, **kwargs)
 
-        Worker.handle_exception = sentry_patched_handle_exception
+        Worker.handle_exception = sentry_patched_handle_exception  # type: ignore[method-assign]
 
         old_enqueue_job = Queue.enqueue_job
 
         @ensure_integration_enabled(SentryIntegration, old_enqueue_job)
-        def sentry_patched_enqueue_job(self, job, **kwargs):
-            # type: (Queue, Any, **Any) -> Any
+        def sentry_patched_enqueue_job(self: Queue, job: Any, **kwargs: Any) -> Any:
             scope = sentry_sdk.get_current_scope()
             if scope.span is not None:
                 job.meta["_sentry_trace_headers"] = dict(
@@ -97,46 +95,30 @@ def sentry_patched_enqueue_job(self, job, **kwargs):
 
             return old_enqueue_job(self, job, **kwargs)
 
-        Queue.enqueue_job = sentry_patched_enqueue_job
+        Queue.enqueue_job = sentry_patched_enqueue_job  # type: ignore[method-assign]
 
         ignore_logger("rq.worker")
 
 
-def _make_event_processor(weak_job):
-    # type: (Callable[[], Job]) -> EventProcessor
-    def event_processor(event, hint):
-        # type: (Event, dict[str, Any]) -> Event
+def _make_event_processor(weak_job: Callable[[], JobModel]) -> EventProcessor:
+    def event_processor(event: Event, hint: dict[str, Any]) -> Event:
         job = weak_job()
         if job is not None:
             with capture_internal_exceptions():
                 extra = event.setdefault("extra", {})
-                rq_job = {
-                    "job_id": job.id,
-                    "func": job.func_name,
-                    "args": job.args,
-                    "kwargs": job.kwargs,
-                    "description": job.description,
-                }
-
-                if job.enqueued_at:
-                    rq_job["enqueued_at"] = format_timestamp(job.enqueued_at)
-                if job.started_at:
-                    rq_job["started_at"] = format_timestamp(job.started_at)
-
-                extra["rq-job"] = rq_job
+                extra["job"] = job.serialize()
 
         if "exc_info" in hint:
             with capture_internal_exceptions():
                 if issubclass(hint["exc_info"][0], JobTimeoutException):
-                    event["fingerprint"] = ["rq", "JobTimeoutException", job.func_name]
+                    event["fingerprint"] = ["django-tasks-scheduler", "JobTimeoutException", job.func_name]
 
         return event
 
     return event_processor
 
 
-def _capture_exception(exc_info, **kwargs):
-    # type: (ExcInfo, **Any) -> None
+def _capture_exception(exc_info: ExcInfo, **kwargs: Any) -> None:
     client = sentry_sdk.get_client()
 
     event, hint = event_from_exception(

scheduler/management/commands/delete_failed_executions.py

@@ -8,7 +8,7 @@
 class Command(BaseCommand):
     help = "Delete failed jobs from Django queue."
 
-    def add_arguments(self, parser):
+    def add_arguments(self, parser: CommandParser) -> None:
         parser.add_argument("--queue", "-q", dest="queue", default="default", help="Specify the queue [default]")
         parser.add_argument("-f", "--func", help='optional job function name, e.g. "app.tasks.func"')
         parser.add_argument("--dry-run", action="store_true", help="Do not actually delete failed jobs")

scheduler/management/commands/export.py

@@ -1,7 +1,8 @@
 import sys
+from typing import Any
 
 import click
-from django.core.management.base import BaseCommand
+from django.core.management.base import BaseCommand, CommandParser
 
 from scheduler.models import Task
 
@@ -11,7 +12,7 @@ class Command(BaseCommand):
 
     help = __doc__
 
-    def add_arguments(self, parser):
+    def add_arguments(self, parser: CommandParser) -> None:
         parser.add_argument(
             "-o",
             "--output",
@@ -37,7 +38,7 @@ def add_arguments(self, parser):
             help="File name to load (otherwise writes to standard output)",
         )
 
-    def handle(self, *args, **options):
+    def handle(self, *args: Any, **options: Any) -> None:
         file = open(options.get("filename"), "w") if options.get("filename") else sys.stdout
         res = list()
 

scheduler/management/commands/import.py

@@ -4,7 +4,7 @@
 import click
 from django.conf import settings
 from django.contrib.contenttypes.models import ContentType
-from django.core.management.base import BaseCommand
+from django.core.management.base import BaseCommand, CommandParser
 from django.utils import timezone
 
 from scheduler.models import TaskArg, TaskKwarg, Task
@@ -24,11 +24,11 @@ def get_task_type(model_str: str) -> TaskType:
     except ValueError:
         pass
     if model_str == "CronTask":
-        return TaskType.CRON
+        return TaskType(TaskType.CRON)
     elif model_str == "RepeatableTask":
-        return TaskType.REPEATABLE
+        return TaskType(TaskType.REPEATABLE)
     elif model_str in {"ScheduledTask", "OnceTask"}:
-        return TaskType.ONCE
+        return TaskType(TaskType.ONCE)
     raise ValueError(f"Invalid model {model_str}")
 
 
@@ -52,12 +52,17 @@ def create_task_from_dict(task_dict: Dict[str, Any], update: bool) -> Optional[T
         if not settings.USE_TZ and not timezone.is_naive(target):
             target = timezone.make_naive(target)
         kwargs["scheduled_time"] = target
-    model_fields = filter(lambda field: hasattr(field, "attname"), Task._meta.get_fields())
-    model_fields = set(map(lambda field: field.attname, model_fields))
+    model_fields = set(map(
+        lambda field: field.attname,
+        filter(
+            lambda field: hasattr(field, "attname"),
+            Task._meta.get_fields()
+        )
+    ))
     keys_to_ignore = list(filter(lambda _k: _k not in model_fields, kwargs.keys()))
     for k in keys_to_ignore:
         del kwargs[k]
-    task = Task.objects.create(**kwargs)
+    task: Task = Task.objects.create(**kwargs)
     click.echo(f"Created task {task}")
     content_type = ContentType.objects.get_for_model(task)
 
@@ -77,13 +82,11 @@ def create_task_from_dict(task_dict: Dict[str, Any], update: bool) -> Optional[T
 
 
 class Command(BaseCommand):
-    """
-    Import scheduled jobs
-    """
+    """Import scheduled jobs"""
 
     help = __doc__
 
-    def add_arguments(self, parser):
+    def add_arguments(self, parser: CommandParser) -> None:
         parser.add_argument(
             "-f",
             "--format",
@@ -115,8 +118,8 @@ def add_arguments(self, parser):
             help="Update existing records",
         )
 
-    def handle(self, *args, **options):
-        file = open(options.get("filename")) if options.get("filename") else sys.stdin
+    def handle(self, *args: Any, **options: Any) -> None:
+        file = open(options.get("filename")) if options.get("filename") else sys.stdin # type: ignore[arg-type]
         jobs = list()
         if options.get("format") == "json":
             import json
@@ -133,11 +136,11 @@ def handle(self, *args, **options):
                 click.echo("Aborting. LibYAML is not installed.")
                 exit(1)
             # Disable YAML alias
-            yaml.Dumper.ignore_aliases = lambda *x: True
+            yaml.Dumper.ignore_aliases = lambda *x: True  # type: ignore[method-assign]
             jobs = yaml.load(file, yaml.SafeLoader)
 
         if options.get("reset"):
             Task.objects.all().delete()
 
         for job in jobs:
-            create_task_from_dict(job, update=options.get("update"))
+            create_task_from_dict(job, update=options.get("update")) # type: ignore[arg-type]

scheduler/management/commands/run_job.py

@@ -1,5 +1,7 @@
+from typing import Any
+
 import click
-from django.core.management.base import BaseCommand
+from django.core.management.base import BaseCommand, CommandParser
 
 from scheduler.helpers.queues import get_queue
 
@@ -13,7 +15,7 @@ class Command(BaseCommand):
     help = __doc__
     args = "<function arg arg ...>"
 
-    def add_arguments(self, parser):
+    def add_arguments(self, parser: CommandParser) -> None:
         parser.add_argument("--queue", "-q", dest="queue", default="default", help="Specify the queue [default]")
         parser.add_argument("--timeout", "-t", type=int, dest="timeout", help="A timeout in seconds")
         parser.add_argument(
@@ -25,7 +27,7 @@ def add_arguments(self, parser):
         )
         parser.add_argument("args", nargs="*", help="Args for callable")
 
-    def handle(self, **options):
+    def handle(self, **options:Any)-> None:
         verbosity = int(options.get("verbosity", 1))
         timeout = options.get("timeout")
         result_ttl = options.get("result_ttl")

scheduler/management/commands/scheduler_stats.py

@@ -3,7 +3,7 @@
 from typing import Any, Dict, List, Optional
 
 import click
-from django.core.management.base import BaseCommand
+from django.core.management.base import BaseCommand, CommandParser
 
 from scheduler.views import get_statistics
 
@@ -24,7 +24,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.table_width = 80
         self.interval = None
 
-    def add_arguments(self, parser: ArgumentParser) -> None:
+    def add_arguments(self, parser: CommandParser) -> None:
         parser.add_argument(
             "-j",
             "--json",

scheduler/views/job_views.py

@@ -21,8 +21,8 @@ class JobDetailAction(str, Enum):
     CANCEL = "cancel"
 
 
-@never_cache
-@staff_member_required
+@never_cache  # type: ignore
+@staff_member_required  # type: ignore
 def job_detail(request: HttpRequest, job_name: str) -> HttpResponse:
     queue, job = _find_job(job_name)
     if job is None or queue is None:
@@ -50,8 +50,8 @@ def job_detail(request: HttpRequest, job_name: str) -> HttpResponse:
     return render(request, "admin/scheduler/job_detail.html", context_data)
 
 
-@never_cache
-@staff_member_required
+@never_cache # type: ignore
+@staff_member_required # type: ignore
 def job_action(request: HttpRequest, job_name: str, action: str) -> HttpResponse:
     queue, job = _find_job(job_name)
     if job is None or queue is None:
@@ -80,6 +80,9 @@ def job_action(request: HttpRequest, job_name: str, action: str) -> HttpResponse
             messages.info(request, f"You have successfully enqueued {job.name}")
             return redirect("job_details", job_name)
         elif action == JobDetailAction.CANCEL:
+            if job.worker_name is None:
+                messages.warning(request, "You cannot cancel a job that has no worker assigned")
+                return redirect("job_details", job_name)
             send_command(
                 connection=queue.connection, command=StopJobCommand(job_name=job.name, worker_name=job.worker_name)
             )