feat: ensure valid task_name before create a Routine

django_cloud_tasks/apps.py

@@ -102,3 +102,6 @@ async def _get_subscriptions():
                     removed.append(subscribed_task)
 
         return updated, removed
+
+    def ready(self):
+        from django_cloud_tasks import signals

django_cloud_tasks/models.py

@@ -100,10 +100,10 @@ def revert(self):
             self.status = self.Statuses.ABORTED
             self.save()
 
+    @property
     def task(self) -> Optional[tasks.Task]:
         app = apps.get_app_config("django_cloud_tasks")
-        tasks = app.on_demand_tasks.copy()
-        return tasks.get(self.task_name)
+        return app.get_task(name=self.task_name)
 
 
 class RoutineVertex(models.Model):

django_cloud_tasks/signals.py

@@ -0,0 +1,13 @@
+from django.core.exceptions import ValidationError
+from django.db.models.signals import pre_save
+from django.dispatch import receiver
+
+from django_cloud_tasks import models, exceptions
+
+
+@receiver(pre_save, sender=models.Routine)
+def ensure_valid_task_name(sender, instance, **kwargs):
+    try:
+        instance.task
+    except exceptions.TaskNotFound:
+        raise ValidationError(f"The task {instance.task_name} was not found. Make sure {instance.task_name} is properly set.")

sample_project/sample_app/tests/tests_models.py

@@ -52,6 +52,11 @@ def tests_revert_not_processed_routine(self):
             self.assertEqual("aborted", routine.status)
         revert_task.assert_not_called()
 
+    def tests_ensure_valid_task_name(self):
+        task_name = "InvalidTaskName"
+        with self.assertRaises(ValidationError, msg=f"The task {task_name} was not found. Make sure {task_name} is properly set."):
+            factories.RoutineFactory(task_name=task_name)
+
 class PipelineModelTest(TestCase):
     def tests_start_pipeline(self):
         pipeline = factories.PipelineFactory()