wip

Author: cunla

docs/changelog.md

@@ -1,5 +1,16 @@
 # Changelog
 
+
+## v2.2.0 🌈
+
+### 🚀 Features
+
+- Created a new `Task` model representing all kind of scheduled tasks. 
+  - In future versions, `CronTask`, `ScheduledTask` and `RepeatableTask` will be removed.
+  - `Task` model has a `task_type` field to differentiate between the types of tasks.
+  - Old tasks in the database will be migrated to the new `Task` model automatically.
+
+
 ## v2.1.0 🌈
 
 ### 🚀 Features

pyproject.toml

@@ -7,7 +7,7 @@ name = "django-tasks-scheduler"
 packages = [
     { include = "scheduler" },
 ]
-version = "2.1.0"
+version = "2.2.0"
 description = "An async job scheduler for django using redis"
 readme = "README.md"
 keywords = ["redis", "django", "background-jobs", "job-queue", "task-queue", "redis-queue", "scheduled-jobs"]

scheduler/management/commands/import.py

@@ -9,6 +9,7 @@
 from django.utils import timezone
 
 from scheduler.models import TaskArg, TaskKwarg, Task
+from scheduler.models.task import TaskType
 from scheduler.tools import MODEL_NAMES
 
 
@@ -17,16 +18,16 @@ def job_model_str(model_str: str) -> str:
         return model_str[:-3] + "Task"
     return model_str
 
-def get_task_type(model_str: str) -> Task.TaskType:
+def get_task_type(model_str: str) -> TaskType:
     model_str = job_model_str(model_str)
     if model_str not in MODEL_NAMES:
         raise ValueError(f"Invalid model {model_str}")
     if model_str == "CronTask":
-        return Task.TaskType.CRON
+        return TaskType.CRON
     elif model_str == "RepeatableTask":
-        return Task.TaskType.REPEATABLE
+        return TaskType.REPEATABLE
     elif model_str == "ScheduledTask":
-        return Task.TaskType.ONCE
+        return TaskType.ONCE
 
 def create_task_from_dict(task_dict: Dict[str, Any], update):
     existing_job = Task.objects.filter(name=task_dict["name"]).first()

scheduler/models/scheduled_task.py

@@ -70,7 +70,7 @@ def get_queue_choices():
 class BaseTask(models.Model):
     created = models.DateTimeField(auto_now_add=True)
     modified = models.DateTimeField(auto_now=True)
-    TASK_TYPE = "BaseTask"
+    task_type = "BaseTask"
     name = models.CharField(
         _("name"),
         max_length=128,
@@ -180,7 +180,7 @@ def _enqueue_args(self) -> Dict:
         """
         res = dict(
             meta=dict(
-                task_type=self.TASK_TYPE,
+                task_type=self.task_type,
                 scheduled_task_id=self.id,
             ),
             on_success=success_callback,
@@ -227,7 +227,7 @@ def schedule(self) -> bool:
         job = self.rqueue.enqueue_at(
             schedule_time,
             tools.run_task,
-            args=(self.TASK_TYPE, self.id),
+            args=(self.task_type, self.id),
             **kwargs,
         )
         self.job_id = job.id
@@ -239,7 +239,7 @@ def enqueue_to_run(self) -> bool:
         kwargs = self._enqueue_args()
         job = self.rqueue.enqueue(
             tools.run_task,
-            args=(self.TASK_TYPE, self.id),
+            args=(self.task_type, self.id),
             **kwargs,
         )
         self.job_id = job.id
@@ -266,7 +266,7 @@ def _schedule_time(self):
     def to_dict(self) -> Dict:
         """Export model to dictionary, so it can be saved as external file backup"""
         res = dict(
-            model=self.TASK_TYPE,
+            model=self.task_type,
             name=self.name,
             callable=self.callable,
             callable_args=[
@@ -312,7 +312,7 @@ def get_absolute_url(self):
 
     def __str__(self):
         func = self.function_string()
-        return f"{self.TASK_TYPE}[{self.name}={func}]"
+        return f"{self.task_type}[{self.name}={func}]"
 
     def save(self, **kwargs):
         schedule_job = kwargs.pop("schedule_job", True)
@@ -391,7 +391,7 @@ class Meta:
 
 
 class ScheduledTask(ScheduledTimeMixin, BaseTask):
-    TASK_TYPE = "ScheduledTask"
+    task_type = "ScheduledTask"
 
     def ready_for_schedule(self) -> bool:
         return super(ScheduledTask, self).ready_for_schedule() and (
@@ -405,6 +405,7 @@ class Meta:
 
 
 class RepeatableTask(RepeatableMixin, ScheduledTimeMixin, BaseTask):
+    task_type = "RepeatableTask"
     class TimeUnits(models.TextChoices):
         SECONDS = "seconds", _("seconds")
         MINUTES = "minutes", _("minutes")
@@ -422,7 +423,6 @@ class TimeUnits(models.TextChoices):
         null=True,
         help_text=_("Number of times to run the job. Leaving this blank means it will run forever."),
     )
-    TASK_TYPE = "RepeatableTask"
 
     def clean(self):
         super(RepeatableTask, self).clean()
@@ -497,7 +497,7 @@ class Meta:
 
 
 class CronTask(RepeatableMixin, BaseTask):
-    TASK_TYPE = "CronTask"
+    task_type = "CronTask"
 
     cron_string = models.CharField(
         _("cron string"),

scheduler/models/task.py

@@ -64,12 +64,12 @@ def get_queue_choices():
     return [(queue, queue) for queue in QUEUES.keys()]
 
 
-class Task(models.Model):
-    class TaskType(models.TextChoices):
-        CRON = "CronTask", _("Cron Task")
-        REPEATABLE = "RepeatableTask", _("Repeatable Task")
-        ONCE = "OnceTask", _("Run once")
+class TaskType(models.TextChoices):
+    CRON = "CronTask", _("Cron Task")
+    REPEATABLE = "RepeatableTask", _("Repeatable Task")
+    ONCE = "OnceTask", _("Run once")
 
+class Task(models.Model):
     class TimeUnits(models.TextChoices):
         SECONDS = "seconds", _("seconds")
         MINUTES = "minutes", _("minutes")
@@ -243,7 +243,7 @@ def _enqueue_args(self) -> Dict:
             res["job_timeout"] = self.timeout
         if self.result_ttl is not None:
             res["result_ttl"] = self.result_ttl
-        if self.task_type == self.TaskType.REPEATABLE:
+        if self.task_type == TaskType.REPEATABLE:
             res["meta"]["interval"] = self.interval_seconds()
             res["meta"]["repeat"] = self.repeat
         return res
@@ -267,7 +267,7 @@ def ready_for_schedule(self) -> bool:
         if not self.enabled:
             logger.debug(f"Task {str(self)} disabled, enable task before scheduling")
             return False
-        if self.task_type == Task.TaskType.REPEATABLE and self._schedule_time() < timezone.now():
+        if self.task_type == TaskType.REPEATABLE and self._schedule_time() < timezone.now():
             return False
         return True
 
@@ -316,9 +316,9 @@ def unschedule(self) -> bool:
         return True
 
     def _schedule_time(self):
-        if self.task_type == self.TaskType.CRON:
+        if self.task_type == TaskType.CRON:
             self.scheduled_time = tools.get_next_cron_time(self.cron_string)
-        elif self.task_type == self.TaskType.REPEATABLE:
+        elif self.task_type == TaskType.REPEATABLE:
             _now = timezone.now()
             if self.scheduled_time >= _now:
                 return utc(self.scheduled_time) if django_settings.USE_TZ else self.scheduled_time
@@ -458,8 +458,8 @@ def clean_cron_string(self):
     def clean(self):
         self.clean_queue()
         self.clean_callable()
-        if self.task_type == self.TaskType.CRON:
+        if self.task_type == TaskType.CRON:
             self.clean_cron_string()
-        if self.task_type == self.TaskType.REPEATABLE:
+        if self.task_type == TaskType.REPEATABLE:
             self.clean_interval_unit()
             self.clean_result_ttl()

scheduler/tests/test_cron_task.py

@@ -3,7 +3,7 @@
 from scheduler import settings
 from scheduler.models import CronTask
 from scheduler.tools import create_worker
-from .test_models import BaseTestCases
+from .test_old_models import BaseTestCases
 from .testtools import task_factory
 from ..queues import get_queue
 

scheduler/tests/test_internals.py

@@ -10,8 +10,8 @@
 class TestInternals(SchedulerBaseCase):
     def test_get_scheduled_job(self):
         task = task_factory(ScheduledTask, scheduled_time=timezone.now() - timedelta(hours=1))
-        self.assertEqual(task, get_scheduled_task(task.TASK_TYPE, task.id))
+        self.assertEqual(task, get_scheduled_task(task.task_type, task.id))
         with self.assertRaises(ValueError):
-            get_scheduled_task(task.TASK_TYPE, task.id + 1)
+            get_scheduled_task(task.task_type, task.id + 1)
         with self.assertRaises(ValueError):
             get_scheduled_task("UNKNOWN_JOBTYPE", task.id)

scheduler/tests/test_mgmt_cmds.py

@@ -7,12 +7,13 @@
 from django.core.management import call_command
 from django.test import TestCase
 
-from scheduler.models import ScheduledTask, RepeatableTask
+from scheduler.models import ScheduledTask, RepeatableTask, Task
 from scheduler.queues import get_queue
 from scheduler.tests.jobs import failing_job, test_job
 from scheduler.tests.testtools import task_factory
 from . import test_settings  # noqa
 from .test_views import BaseTestCase
+from ..models.task import TaskType
 from ..tools import create_worker
 
 
@@ -231,8 +232,8 @@ def test_import__should_schedule_job(self):
         # act
         call_command("import", filename=self.tmpfile.name)
         # assert
-        self.assertEqual(1, ScheduledTask.objects.count())
-        db_job = ScheduledTask.objects.first()
+        self.assertEqual(1, Task.objects.filter(task_type=TaskType.ONCE).count())
+        db_job = Task.objects.filter(task_type=TaskType.ONCE).first()
         attrs = ["name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
             self.assertEqual(getattr(jobs[0], attr), getattr(db_job, attr))
@@ -247,8 +248,8 @@ def test_import__should_schedule_job_yaml(self):
         # act
         call_command("import", filename=self.tmpfile.name, format="yaml")
         # assert
-        self.assertEqual(1, ScheduledTask.objects.count())
-        db_job = ScheduledTask.objects.first()
+        self.assertEqual(1, Task.objects.filter(task_type=TaskType.ONCE).count())
+        db_job = Task.objects.filter(task_type=TaskType.ONCE).objects.first()
         attrs = ["name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
             self.assertEqual(getattr(jobs[0], attr), getattr(db_job, attr))
@@ -282,13 +283,13 @@ def test_import__should_schedule_job_reset(self):
             reset=True,
         )
         # assert
-        self.assertEqual(1, ScheduledTask.objects.count())
-        db_job = ScheduledTask.objects.first()
+        self.assertEqual(1, Task.objects.filter(task_type=TaskType.ONCE).count())
+        db_job = Task.objects.filter(task_type=TaskType.ONCE).first()
         attrs = ["name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
             self.assertEqual(getattr(jobs[0], attr), getattr(db_job, attr))
-        self.assertEqual(1, RepeatableTask.objects.count())
-        db_job = RepeatableTask.objects.first()
+        self.assertEqual(1, Task.objects.filter(task_type=TaskType.REPEATABLE).count())
+        db_job = Task.objects.filter(task_type=TaskType.REPEATABLE).first()
         attrs = ["name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
             self.assertEqual(getattr(jobs[1], attr), getattr(db_job, attr))
@@ -307,8 +308,8 @@ def test_import__should_schedule_job_update_existing(self):
             update=True,
         )
         # assert
-        self.assertEqual(2, ScheduledTask.objects.count())
-        db_job = ScheduledTask.objects.get(name=jobs[0].name)
+        self.assertEqual(2, Task.objects.filter(task_type=TaskType.ONCE).count())
+        db_job = Task.objects.filter(task_type=TaskType.ONCE).get(name=jobs[0].name)
         self.assertNotEqual(jobs[0].id, db_job.id)
         attrs = ["name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
@@ -327,8 +328,8 @@ def test_import__should_schedule_job_without_update_existing(self):
             filename=self.tmpfile.name,
         )
         # assert
-        self.assertEqual(2, ScheduledTask.objects.count())
-        db_job = ScheduledTask.objects.get(name=jobs[0].name)
+        self.assertEqual(2, Task.objects.filter(task_type=TaskType.ONCE).count())
+        db_job = Task.objects.get(name=jobs[0].name)
         attrs = ["id", "name", "queue", "callable", "enabled", "timeout"]
         for attr in attrs:
             self.assertEqual(getattr(jobs[0], attr), getattr(db_job, attr))

scheduler/tests/test_models.py renamed to scheduler/tests/test_old_models.py

@@ -376,9 +376,7 @@ def test_admin_change_view__bad_redis_connection(self):
         def test_admin_enqueue_job_now(self):
             # arrange
             self.client.login(username="admin", password="admin")
-            task = task_factory(
-                self.TaskModelClass,
-            )
+            task = task_factory(self.TaskModelClass)
             self.assertIsNotNone(task.job_id)
             self.assertTrue(task.is_scheduled())
             data = {
@@ -396,7 +394,7 @@ def test_admin_enqueue_job_now(self):
             self.assertEqual(200, res.status_code)
             entry = _get_job_from_scheduled_registry(task)
             task_model, scheduled_task_id = entry.args
-            self.assertEqual(task_model, task.TASK_TYPE)
+            self.assertEqual(task_model, task.task_type)
             self.assertEqual(scheduled_task_id, task.id)
             self.assertEqual("scheduled", entry.get_status())
             assert_has_execution_with_status(task, "queued")
@@ -410,7 +408,7 @@ def test_admin_enqueue_job_now(self):
 
             # assert 2
             entry = _get_job_from_scheduled_registry(task)
-            self.assertEqual(task_model, task.TASK_TYPE)
+            self.assertEqual(task_model, task.task_type)
             self.assertEqual(scheduled_task_id, task.id)
             assert_has_execution_with_status(task, "finished")
 

scheduler/tests/test_repeatable_task.py

@@ -6,7 +6,7 @@
 
 from scheduler import settings
 from scheduler.models import RepeatableTask
-from scheduler.tests.test_models import BaseTestCases
+from scheduler.tests.test_old_models import BaseTestCases
 from .testtools import task_factory, _get_job_from_scheduled_registry