Post execute signal and tests (#580)

* adds post_execute signal and tests

* Updates signal docts

* Fixes typo in docs

django_q/cluster.py

@@ -43,7 +43,7 @@ from django_q.conf import (
 from django_q.humanhash import humanize
 from django_q.models import Schedule, Success, Task
 from django_q.queues import Queue
-from django_q.signals import pre_execute
+from django_q.signals import post_execute, pre_execute
 from django_q.signing import BadSignature, SignedPackage
 from django_q.status import Stat, Status
 
@@ -386,6 +386,8 @@ def monitor(result_queue: Queue, broker: Broker = None):
         ack_id = task.pop("ack_id", False)
         if ack_id and (task["success"] or task.get("ack_failure", False)):
             broker.acknowledge(ack_id)
+        # signal execution done
+        post_execute.send(sender="django_q", task=task)
         # log the result
         if task["success"]:
             # log success
@@ -745,7 +747,7 @@ def rss_check():
 
 
 def localtime() -> datetime:
-    """ Override for timezone.localtime to deal with naive times and local times"""
+    """Override for timezone.localtime to deal with naive times and local times"""
     if settings.USE_TZ:
         return timezone.localtime()
     return datetime.now()

django_q/signals.py

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

django_q/tests/test_cluster.py

@@ -2,8 +2,10 @@ import os
 import sys
 import threading
 import uuid as uuidlib
+from math import copysign
 from multiprocessing import Event, Value
 from time import sleep
+from typing import Optional
 
 import pytest
 from django.utils import timezone
@@ -17,6 +19,7 @@ from django_q.conf import Conf
 from django_q.humanhash import DEFAULT_WORDLIST, uuid
 from django_q.models import Success, Task
 from django_q.queues import Queue
+from django_q.signals import post_execute, pre_enqueue, pre_execute
 from django_q.status import Stat
 from django_q.tasks import (
     async_task,
@@ -609,6 +612,85 @@ def test_acknowledge_failure_override():
     assert broker.acknowledgements.get("test_success_ack_id") == 1
 
 
+class TestSignals:
+    @pytest.mark.django_db
+    def test_pre_enqueue_signal(self, broker):
+        broker.list_key = "pre_enqueue_test:q"
+        broker.delete_queue()
+        self.signal_was_called: bool = False
+        self.task: Optional[dict] = None
+
+        def handler(sender, task, **kwargs):
+            self.signal_was_called = True
+            self.task = task
+
+        pre_enqueue.connect(handler)
+        task_id = async_task("math.copysign", 1, -1, broker=broker)
+        assert self.signal_was_called is True
+        assert self.task.get("id") == task_id
+        pre_enqueue.disconnect(handler)
+        broker.delete_queue()
+
+    @pytest.mark.django_db
+    def test_pre_execute_signal(self, broker):
+        broker.list_key = "pre_execute_test:q"
+        broker.delete_queue()
+        self.signal_was_called: bool = False
+        self.task: Optional[dict] = None
+        self.func = None
+
+        def handler(sender, task, func, **kwargs):
+            self.signal_was_called = True
+            self.task = task
+            self.func = func
+
+        pre_execute.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.func == copysign
+        pre_execute.disconnect(handler)
+
+    @pytest.mark.django_db
+    def test_post_execute_signal(self, broker):
+        broker.list_key = "post_execute_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.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.disconnect(handler)
+
+
 @pytest.mark.django_db
 def assert_result(task):
     assert task is not None

docs/signals.rst

@@ -24,20 +24,31 @@ executed by a worker. This signal provides two arguments:
   with a function path, this argument will be the callable function
   nonetheless.
 
+After executing a task
+""""""""""""""""""""""
+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
 -----------------------
 
-Connecting to a Django Q signal is done in the same manner as any other Django
+Connecting to a Django Q signal is done the same as any other Django
 signal::
 
     from django.dispatch import receiver
-    from django_q.signals import pre_enqueue, pre_execute
+    from django_q.signals import pre_enqueue, pre_execute, post_execute
 
     @receiver(pre_enqueue)
     def my_pre_enqueue_callback(sender, task, **kwargs):
-        print("Task {} will be enqueued".format(task["name"]))
+        print(f"Task {task['name']} will be queued")
 
     @receiver(pre_execute)
     def my_pre_execute_callback(sender, func, task, **kwargs):
-        print("Task {} will be executed by calling {}".format(
-              task["name"], func))
+        print(f"Task {task['name']} will be executed by calling {func}")
+
+    @receiver(post_execute)
+    def my_post_execute_callback(sender, task, **kwargs):
+        print(f"Task {task['name']} was executed with result {task['result']}")
+
+