diff --git a/django_q/signals.py b/django_q/signals.py index 1476089..d9af1fa 100644 --- a/django_q/signals.py +++ b/django_q/signals.py @@ -43,3 +43,6 @@ pre_execute = Signal() # args: task post_execute = Signal() + +# args: func, task +post_execute_in_worker = Signal() diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index b61b0db..6065a7a 100644 --- a/django_q/tests/test_cluster.py +++ b/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): diff --git a/django_q/worker.py b/django_q/worker.py index 91d45e4..1fd87f2 100644 --- a/django_q/worker.py +++ b/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 diff --git a/docs/signals.rst b/docs/signals.rst index 601d8e5..51a1354 100644 --- a/docs/signals.rst +++ b/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 -executed by a worker and processed by the monitor. It included the ``task`` dictionary with the result. +- The ``django_q.signals.post_execute_in_worker`` signal is emitted after a task + 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 @@ -56,6 +62,10 @@ signal:: @receiver(post_execute) 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):