Allow jobs to be defined as coroutines

django_dbq/management/commands/worker.py

@@ -6,6 +6,7 @@
 from simplesignals.process import WorkerProcessBase
 from time import sleep
 import logging
+from django_dbq.utils import run_job
 
 
 logger = logging.getLogger(__name__)
@@ -35,7 +36,7 @@ def process_job(queue_name):
 
     try:
         task_function = import_string(job.next_task)
-        task_function(job)
+        run_job(task_function, job)
         job.update_next_task()
         if not job.next_task:
             job.state = Job.STATES.COMPLETE
@@ -51,7 +52,7 @@ def process_job(queue_name):
                 "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)
+            run_job(failure_hook_function, job, exception)
         else:
             logger.info("No failure hook for job id=%s", job.pk)
 

django_dbq/models.py

@@ -11,6 +11,7 @@
 import datetime
 import logging
 import uuid
+from django_dbq.utils import run_job
 
 
 logger = logging.getLogger(__name__)
@@ -137,7 +138,7 @@ def run_creation_hook(self):
         if creation_hook_name:
             logger.info("Running creation hook %s for new job", creation_hook_name)
             creation_hook_function = import_string(creation_hook_name)
-            creation_hook_function(self)
+            run_job(creation_hook_function, self)
 
     @staticmethod
     def get_queue_depths():

django_dbq/utils.py

@@ -0,0 +1,8 @@
+import asyncio
+
+
+def run_job(job_function, *args, **kwargs):
+    if asyncio.iscoroutinefunction(job_function):
+        asyncio.run(job_function(*args, **kwargs))
+    else:
+        job_function(*args, **kwargs)