Rework where pre and post task hooks are run, and fix tests

Author: j4mie

django_dbq/management/commands/worker.py

@@ -61,8 +61,6 @@ def _process_job(self):
             if not job:
                 return
 
-            job.run_pre_task_hook()
-
             logger.info(
                 'Processing job: name="%s" queue="%s" id=%s state=%s next_task=%s',
                 job.name,
@@ -77,7 +75,10 @@ def _process_job(self):
 
         try:
             task_function = import_string(job.next_task)
+
+            job.run_pre_task_hook()
             task_function(job)
+
             job.update_next_task()
             if not job.next_task:
                 job.state = Job.STATES.COMPLETE
@@ -96,6 +97,11 @@ def _process_job(self):
                 failure_hook_function(job, exception)
             else:
                 logger.info("No failure hook for job id=%s", job.pk)
+        finally:
+            try:
+                job.run_post_task_hook()
+            except:
+                logger.exception("Job id=%s post_task_hook failed", job.pk)
 
         logger.info(
             'Updating job: name="%s" id=%s state=%s next_task=%s',
@@ -111,8 +117,6 @@ def _process_job(self):
             logger.exception("Failed to save job: id=%s", job.pk)
             raise
 
-        job.run_post_task_hook()
-
         self.current_job = None
 
 

django_dbq/tests.py

@@ -343,7 +343,7 @@ def test_creation_hook_only_runs_on_create(self):
 @override_settings(
     JOBS={
         "testjob": {
-            "tasks": ["django_dbq.tests.failing_task"],
+            "tasks": ["django_dbq.tests.test_task"],
             "pre_task_hook": "django_dbq.tests.pre_task_hook",
         }
     }
@@ -353,15 +353,15 @@ def test_pre_task_hook(self):
         job = Job.objects.create(name="testjob")
         Worker("default", 1)._process_job()
         job = Job.objects.get()
-        self.assertEqual(job.state, Job.STATES.FAILED)
-        self.assertEqual(job.workspace["output"], "failure hook ran")
+        self.assertEqual(job.state, Job.STATES.COMPLETE)
+        self.assertEqual(job.workspace["output"], "pre task hook ran")
         self.assertEqual(job.workspace["job_id"], str(job.id))
 
 
 @override_settings(
     JOBS={
         "testjob": {
-            "tasks": ["django_dbq.tests.failing_task"],
+            "tasks": ["django_dbq.tests.test_task"],
             "post_task_hook": "django_dbq.tests.post_task_hook",
         }
     }
@@ -371,7 +371,7 @@ def test_post_task_hook(self):
         job = Job.objects.create(name="testjob")
         Worker("default", 1)._process_job()
         job = Job.objects.get()
-        self.assertEqual(job.state, Job.STATES.FAILED)
+        self.assertEqual(job.state, Job.STATES.COMPLETE)
         self.assertEqual(job.workspace["output"], "post task hook ran")
         self.assertEqual(job.workspace["job_id"], str(job.id))