diff --git a/django_q/cluster.py b/django_q/cluster.py index 326d735..9bce4f5 100644 --- a/django_q/cluster.py +++ b/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() diff --git a/django_q/signals.py b/django_q/signals.py index 7502c3b..109b473 100644 --- a/django_q/signals.py +++ b/django_q/signals.py @@ -37,3 +37,6 @@ pre_enqueue = Signal() # args: func, task pre_execute = Signal() + +# args: task +post_execute = Signal() diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 5d92177..1755932 100644 --- a/django_q/tests/test_cluster.py +++ b/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 diff --git a/docs/signals.rst b/docs/signals.rst index 41ad08c..7d27041 100644 --- a/docs/signals.rst +++ b/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']}") + +