Refactor to pull task and hook running into Job model

Author: j4mie

django_dbq/management/commands/worker.py

@@ -74,29 +74,18 @@ def _process_job(self):
             self.current_job = job
 
         try:
-            task_function = import_string(job.next_task)
-
             job.run_pre_task_hook()
-            task_function(job)
-
+            job.run_next_task()
             job.update_next_task()
+
             if not job.next_task:
                 job.state = Job.STATES.COMPLETE
             else:
                 job.state = Job.STATES.READY
         except Exception as exception:
             logger.exception("Job id=%s failed", job.pk)
             job.state = Job.STATES.FAILED
-
-            failure_hook_name = job.get_failure_hook_name()
-            if failure_hook_name:
-                logger.info(
-                    "Running failure hook %s for job id=%s", failure_hook_name, job.pk
-                )
-                failure_hook_function = import_string(failure_hook_name)
-                failure_hook_function(job, exception)
-            else:
-                logger.info("No failure hook for job id=%s", job.pk)
+            job.run_failure_hook(exception)
         finally:
             try:
                 job.run_post_task_hook()

django_dbq/models.py

@@ -128,6 +128,10 @@ def save(self, *args, **kwargs):
     def update_next_task(self):
         self.next_task = get_next_task_name(self.name, self.next_task) or ""
 
+    def run_next_task(self):
+        next_task_function = import_string(self.next_task)
+        next_task_function(self)
+
     def get_pre_task_hook_name(self):
         return get_pre_task_hook_name(self.name)
 
@@ -143,21 +147,28 @@ def get_creation_hook_name(self):
     def run_pre_task_hook(self):
         pre_task_hook_name = self.get_pre_task_hook_name()
         if pre_task_hook_name:
-            logger.info("Running pre_task hook %s for new job", pre_task_hook_name)
+            logger.info("Running pre_task hook %s for job", pre_task_hook_name)
             pre_task_hook_function = import_string(pre_task_hook_name)
             pre_task_hook_function(self)
 
     def run_post_task_hook(self):
         post_task_hook_name = self.get_post_task_hook_name()
         if post_task_hook_name:
-            logger.info("Running post_task hook %s for new job", post_task_hook_name)
+            logger.info("Running post_task hook %s for job", post_task_hook_name)
             post_task_hook_function = import_string(post_task_hook_name)
             post_task_hook_function(self)
 
+    def run_failure_hook(self, exception):
+        failure_hook_name = self.get_failure_hook_name()
+        if failure_hook_name:
+            logger.info("Running failure hook %s for job", failure_hook_name)
+            failure_hook_function = import_string(failure_hook_name)
+            failure_hook_function(self, exception)
+
     def run_creation_hook(self):
         creation_hook_name = self.get_creation_hook_name()
         if creation_hook_name:
-            logger.info("Running creation hook %s for new job", creation_hook_name)
+            logger.info("Running creation hook %s for job", creation_hook_name)
             creation_hook_function = import_string(creation_hook_name)
             creation_hook_function(self)