job => task

Author: cunla

scheduler/admin/old_task_models.py

@@ -9,7 +9,7 @@
 
 from scheduler import tools
 from scheduler.broker_types import ConnectionErrorTypes
-from scheduler.models import CronTask, TaskArg, TaskKwarg, RepeatableTask, ScheduledTask, BaseTask
+from scheduler.models import CronTask, TaskArg, TaskKwarg, RepeatableTask, ScheduledTask, BaseTask, migrate_util
 from scheduler.settings import SCHEDULER_CONFIG, logger
 from scheduler.tools import get_job_executions_for_task
 
@@ -113,6 +113,11 @@ class JobKwargInline(HiddenMixin, GenericStackedInline):
 )
 
 
+def get_message_bit(rows_updated: int) -> str:
+    message_bit = "1 task was" if rows_updated == 1 else f"{rows_updated} tasks were"
+    return message_bit
+
+
 @admin.register(CronTask, ScheduledTask, RepeatableTask)
 class TaskAdmin(admin.ModelAdmin):
     """TaskAdmin admin view for all task models.
@@ -230,6 +235,16 @@ def delete_model(self, request, obj):
         obj.unschedule()
         super(TaskAdmin, self).delete_model(request, obj)
 
+    @admin.action(description=_("Migrate to new Task model(s)"), permissions=("change",))
+    def migrate(self, request, queryset):
+        rows_updated = 0
+        for obj in queryset.filter(enabled=True).iterator():
+            migrate_util.migrate(obj)
+            rows_updated += 1
+
+        level = messages.WARNING if not rows_updated else messages.INFO
+        self.message_user(request, f"{get_message_bit(rows_updated)} successfully migrated to new model.", level=level)
+
     @admin.action(description=_("Disable selected %(verbose_name_plural)s"), permissions=("change",))
     def disable_selected(self, request, queryset):
         rows_updated = 0
@@ -238,10 +253,9 @@ def disable_selected(self, request, queryset):
             obj.unschedule()
             rows_updated += 1
 
-        message_bit = "1 job was" if rows_updated == 1 else f"{rows_updated} jobs were"
-
         level = messages.WARNING if not rows_updated else messages.INFO
-        self.message_user(request, f"{message_bit} successfully disabled and unscheduled.", level=level)
+        self.message_user(request, f"{get_message_bit(rows_updated)} successfully disabled and unscheduled.",
+                          level=level)
 
     @admin.action(description=_("Enable selected %(verbose_name_plural)s"), permissions=("change",))
     def enable_selected(self, request, queryset):
@@ -251,9 +265,8 @@ def enable_selected(self, request, queryset):
             obj.save()
             rows_updated += 1
 
-        message_bit = "1 job was" if rows_updated == 1 else f"{rows_updated} jobs were"
         level = messages.WARNING if not rows_updated else messages.INFO
-        self.message_user(request, f"{message_bit} successfully enabled and scheduled.", level=level)
+        self.message_user(request, f"{get_message_bit(rows_updated)} successfully enabled and scheduled.", level=level)
 
     @admin.action(description="Enqueue now", permissions=("change",))
     def enqueue_job_now(self, request, queryset):

scheduler/admin/task_admin.py

@@ -21,6 +21,11 @@ class JobKwargInline(GenericStackedInline):
     fieldsets = ((None, dict(fields=("key", ("arg_type", "val")))),)
 
 
+def get_message_bit(rows_updated: int) -> str:
+    message_bit = "1 task was" if rows_updated == 1 else f"{rows_updated} tasks were"
+    return message_bit
+
+
 @admin.register(Task)
 class TaskAdmin(admin.ModelAdmin):
     """TaskAdmin admin view for all task models."""
@@ -153,10 +158,9 @@ def disable_selected(self, request, queryset):
             obj.unschedule()
             rows_updated += 1
 
-        message_bit = "1 job was" if rows_updated == 1 else f"{rows_updated} jobs were"
-
         level = messages.WARNING if not rows_updated else messages.INFO
-        self.message_user(request, f"{message_bit} successfully disabled and unscheduled.", level=level)
+        self.message_user(request, f"{get_message_bit(rows_updated)} successfully disabled and unscheduled.",
+                          level=level)
 
     @admin.action(description=_("Enable selected %(verbose_name_plural)s"), permissions=("change",))
     def enable_selected(self, request, queryset):
@@ -166,9 +170,8 @@ def enable_selected(self, request, queryset):
             obj.save()
             rows_updated += 1
 
-        message_bit = "1 job was" if rows_updated == 1 else f"{rows_updated} jobs were"
         level = messages.WARNING if not rows_updated else messages.INFO
-        self.message_user(request, f"{message_bit} successfully enabled and scheduled.", level=level)
+        self.message_user(request, f"{get_message_bit(rows_updated)} successfully enabled and scheduled.", level=level)
 
     @admin.action(description="Enqueue now", permissions=("change",))
     def enqueue_job_now(self, request, queryset):

scheduler/models/migrate_util.py

@@ -0,0 +1,87 @@
+from datetime import datetime
+from typing import Dict, Any, Optional
+
+from django.conf import settings
+from django.contrib.contenttypes.models import ContentType
+from django.utils import timezone
+
+from scheduler.models.old_scheduled_task import BaseTask
+from scheduler.models.task import Task, TaskArg, TaskKwarg
+from scheduler.settings import logger
+from scheduler.tools import TaskType
+
+
+def job_model_str(model_str: str) -> str:
+    if model_str.find("Job") == len(model_str) - 3:
+        return model_str[:-3] + "Task"
+    return model_str
+
+
+def get_task_type(model_str: str) -> TaskType:
+    model_str = job_model_str(model_str)
+    if TaskType(model_str):
+        return TaskType(model_str)
+    if model_str == "CronTask":
+        return TaskType.CRON
+    elif model_str == "RepeatableTask":
+        return TaskType.REPEATABLE
+    elif model_str in {"ScheduledTask", "OnceTask"}:
+        return TaskType.ONCE
+    raise ValueError(f"Invalid model {model_str}")
+
+
+def create_task_from_dict(task_dict: Dict[str, Any], recreate: bool) -> Optional[Task]:
+    if "new_task_id" in task_dict:
+        existing_task = Task.objects.filter(id=task_dict["new_task_id"]).first()
+    else:
+        existing_task = Task.objects.filter(name=task_dict["name"]).first()
+    task_type = get_task_type(task_dict["model"])
+    if existing_task:
+        if recreate:
+            logger.info(f'Found existing job "{existing_task}, removing it to be reinserted"')
+            existing_task.delete()
+        else:
+            logger.info(f'Found existing job "{existing_task}", skipping')
+            return None
+    kwargs = dict(task_dict)
+    kwargs["task_type"] = task_type
+    del kwargs["model"]
+    del kwargs["callable_args"]
+    del kwargs["callable_kwargs"]
+    del kwargs["new_task_id"]
+    if kwargs.get("scheduled_time", None):
+        target = datetime.fromisoformat(kwargs["scheduled_time"])
+        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))
+    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)
+    logger.info(f"Created task {task}")
+    content_type = ContentType.objects.get_for_model(task)
+
+    for arg in task_dict["callable_args"]:
+        TaskArg.objects.create(
+            content_type=content_type,
+            object_id=task.id,
+            **arg,
+        )
+    for arg in task_dict["callable_kwargs"]:
+        TaskKwarg.objects.create(
+            content_type=content_type,
+            object_id=task.id,
+            **arg,
+        )
+    return task
+
+
+def migrate(old: BaseTask) -> Optional[Task]:
+    old_task_dict = old.to_dict()
+    new_task = create_task_from_dict(old_task_dict, old_task_dict.get("new_task_id") is not None)
+    old.new_task_id = new_task.id
+    old.enabled = False
+    old.save()
+    return new_task

scheduler/models/old_scheduled_task.py

@@ -299,6 +299,7 @@ def to_dict(self) -> Dict:
             failed_runs=getattr(self, "failed_runs", 0),
             last_successful_run=getattr(self, "last_successful_run", None),
             last_failed_run=getattr(self, "last_failed_run", None),
+            new_task_id=getattr(self, "new_task_id", None),
         )
         return res
 
@@ -396,7 +397,7 @@ class ScheduledTask(ScheduledTimeMixin, BaseTask):
 
     def ready_for_schedule(self) -> bool:
         return super(ScheduledTask, self).ready_for_schedule() and (
-            self.scheduled_time is None or self.scheduled_time >= timezone.now()
+                self.scheduled_time is None or self.scheduled_time >= timezone.now()
         )
 
     class Meta:

scheduler/tests/test_old_models/test_old_models.py

@@ -426,7 +426,7 @@ def test_admin_enable_job(self):
             task.refresh_from_db()
             self.assertTrue(task.enabled)
             self.assertTrue(task.is_scheduled())
-            assert_response_has_msg(res, "1 job was successfully enabled and scheduled.")
+            assert_response_has_msg(res, "1 task was successfully enabled and scheduled.")
 
         def test_admin_disable_job(self):
             # arrange
@@ -449,7 +449,7 @@ def test_admin_disable_job(self):
             task.refresh_from_db()
             self.assertFalse(task.is_scheduled())
             self.assertFalse(task.enabled)
-            assert_response_has_msg(res, "1 job was successfully disabled and unscheduled.")
+            assert_response_has_msg(res, "1 task was successfully disabled and unscheduled.")
 
         def test_admin_single_delete(self):
             # arrange

scheduler/tests/test_old_models/test_old_task_model.py

@@ -428,7 +428,7 @@ def test_admin_enable_job(self):
             task.refresh_from_db()
             self.assertTrue(task.enabled)
             self.assertTrue(task.is_scheduled())
-            assert_response_has_msg(res, "1 job was successfully enabled and scheduled.")
+            assert_response_has_msg(res, "1 task was successfully enabled and scheduled.")
 
         def test_admin_disable_job(self):
             # arrange
@@ -451,7 +451,7 @@ def test_admin_disable_job(self):
             task.refresh_from_db()
             self.assertFalse(task.is_scheduled())
             self.assertFalse(task.enabled)
-            assert_response_has_msg(res, "1 job was successfully disabled and unscheduled.")
+            assert_response_has_msg(res, "1 task was successfully disabled and unscheduled.")
 
         def test_admin_single_delete(self):
             # arrange