Add post_execute_in_worker signal (#309)

django_q/signals.py

@@ -43,3 +43,6 @@ pre_execute = Signal()
 
 # args: task
 post_execute = Signal()
+
+# args: func, task
+post_execute_in_worker = Signal()

django_q/tests/test_cluster.py

@@ -19,7 +19,12 @@ from django_q.models import Success, Task
 from django_q.monitor import monitor, save_task
 from django_q.pusher import pusher
 from django_q.queues import Queue
-from django_q.signals import post_execute, pre_enqueue, pre_execute
+from django_q.signals import (
+    post_execute,
+    post_execute_in_worker,
+    pre_enqueue,
+    pre_execute,
+)
 from django_q.status import Stat
 from django_q.tasks import (
     async_task,
@@ -760,6 +765,35 @@ class TestSignals:
         assert self.task.get("result") == -1
         post_execute.disconnect(handler)
 
+    @pytest.mark.django_db
+    def test_post_execute_in_worker_signal(self, broker):
+        broker.list_key = "post_execute_in_worker_test:q"
+        broker.delete_queue()
+        self.signal_was_called: bool = False
+        self.task: Optional[dict] = None
+        self.func = None
+
+        def handler(sender, task, **kwargs):
+            self.signal_was_called = True
+            self.task = task
+
+        post_execute_in_worker.connect(handler)
+        task_id = async_task("math.copysign", 1, -1, broker=broker)
+        task_queue = Queue()
+        result_queue = Queue()
+        event = Event()
+        event.set()
+        pusher(task_queue, event, broker=broker)
+        task_queue.put("STOP")
+        worker(task_queue, result_queue, Value("f", -1))
+        result_queue.put("STOP")
+        monitor(result_queue, broker)
+        broker.delete_queue()
+        assert self.signal_was_called is True
+        assert self.task.get("id") == task_id
+        assert self.task.get("result") == -1
+        post_execute_in_worker.disconnect(handler)
+
 
 @pytest.mark.django_db
 def assert_result(task):

django_q/worker.py

@@ -18,7 +18,7 @@ except core.exceptions.AppRegistryNotReady:
 
 from django_q.conf import Conf, error_reporter, logger, resource, setproctitle
 from django_q.exceptions import TimeoutException
-from django_q.signals import post_spawn, pre_execute
+from django_q.signals import post_execute_in_worker, post_spawn, pre_execute
 from django_q.timeout import TimeoutHandler
 from django_q.utils import close_old_django_connections, get_func_repr
 
@@ -109,6 +109,7 @@ def worker(
             if error_reporter:
                 error_reporter.report()
             if task.get("sync", False):
+                post_execute_in_worker.send(sender="django_q", func=f, task=task)
                 raise
 
         with timer.get_lock():
@@ -116,6 +117,7 @@ def worker(
             task["result"] = result[0]
             task["success"] = result[1]
             task["stopped"] = timezone.now()
+            post_execute_in_worker.send(sender="django_q", func=f, task=task)
             result_queue.put(task)
             if timeout_error:
                 # force destroy process due to timeout

docs/signals.rst

@@ -32,8 +32,14 @@ executed by a worker. This signal provides two arguments:
 
 After executing a task
 """"""""""""""""""""""
-The ``django_q.signals.post_execute`` signal is emitted after a task is
+- The ``django_q.signals.post_execute_in_worker`` signal is emitted after a task
-executed by a worker and processed by the monitor. It included the ``task`` dictionary with the result.
+  is executed by a worker and processed by the **worker**. It included the ``task``
+  dictionary with the result. Note that this signal is **emitted from, and handled
+  by, the worker process itself**, not the monitor, unlike the ``post_execute``
+  signal below.
+- The ``django_q.signals.post_execute`` signal is emitted after a task is
+  executed by a worker and processed by the **monitor**. It included the ``task``
+  dictionary with the result.
 
 
 Subscribing to a signal
@@ -57,6 +63,10 @@ signal::
     def my_post_execute_callback(sender, task, **kwargs):
         print(f"Task {task['name']} was executed with result {task['result']}")
     
+    @receiver(post_execute_in_worker)
+    def my_post_execute_in_worker_callback(sender, func, task, **kwargs):
+        print(f"Task {task['name']} was executed with result {task['result']}")
+
     @receiver(post_spawn)
     def my_post_spawn_callback(sender, proc_name, **kwargs):
         print(f"Process {proc_name} has spawned")