Release v1.6.0

* Correct health check import in examples.rst

* Update monitor.rst

---------

Co-authored-by: Stan Triepels <1939656+GDay@users.noreply.github.com>

.github/workflows/test.yml

@@ -11,8 +11,25 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [ "3.8", "3.9", "3.10", "3.11" ]
+        python-version: [ "3.8", "3.9", "3.10", "3.11", "3.12.0-rc.3" ]
-        django: [ "3.2", "4.1", "4.2" ]
+        django: [ "3.2", "4.1", "4.2", "5.0a1"]
+        exclude:
+          # django 5.0 does not support 3.8 and 3.9
+          - python-version: "3.8"
+            django: "5.0a1" 
+          - python-version: "3.9"
+            django: "5.0a1" 
+          # django 4.2 does not support 3.12
+          - python-version: "3.12.0-rc.3"
+            django: "4.2" 
+          # django 4.1 does not support 3.12
+          - python-version: "3.12.0-rc.3"
+            django: "4.1" 
+          # django 3.2 does not support 3.11 and 3.12
+          - python-version: "3.11"
+            django: "3.2" 
+          - python-version: "3.12.0-rc.3"
+            django: "3.2" 
 
     services:
       disque:
@@ -39,16 +56,16 @@ jobs:
           - 6379:6379
         options: --entrypoint redis-server
     steps:
-        - uses: actions/checkout@v2
+        - uses: actions/checkout@v3
         - name: Set up Python ${{ matrix.python-version }}
-          uses: actions/setup-python@v2
+          uses: actions/setup-python@v4
           with:
             python-version: ${{ matrix.python-version }}
         - name: Install dependencies with Django ${{ matrix.django }}
           run: |
             python -m pip install --upgrade pip
             pip install poetry
-            poetry add "django==${{ matrix.django }}"
+            poetry add "django==${{ matrix.django }}" --python=${{ matrix.python-version }}
             poetry install -E testing
         - name: Run Tests
           run: |

CHANGELOG.md

@@ -1,6 +1,38 @@
 # Changelog
 
-## [Unreleased](https://github.com/GDay/django-q2/tree/HEAD)
+## [v1.6.0](https://github.com/django-q2/django-q2/tree/v1.6.0) (2023-10-12)
+
+**Merged pull requests:**
+
+- Add support for Django 5 and python 12 https://github.com/django-q2/django-q2/pull/120
+- Fix for "apps not ready" in Windows and Mac https://github.com/django-q2/django-q2/pull/116
+- Update broken MongoClient link in Docs https://github.com/django-q2/django-q2/pull/127
+- Fix German Translation Typo https://github.com/django-q2/django-q2/pull/124
+- Update Add-ons install command in install.rst https://github.com/django-q2/django-q2/pull/115
+- DOCS: Correct health check import in examples.rst https://github.com/django-q2/django-q2/pull/110
+
+## [v1.5.5](https://github.com/django-q2/django-q2/tree/v1.5.5) (2023-09-01)
+
+**Merged pull requests:**
+
+- Add documentation to migrate from django-q to django-q2 https://github.com/django-q2/django-q2/pull/108
+- Fix not picking up result from falsy result https://github.com/django-q2/django-q2/pull/107
+- Remove deprecated usage pkg_resources https://github.com/django-q2/django-q2/pull/103
+- Move worker, scheduler, pusher and monitor to separate files https://github.com/django-q2/django-q2/pull/100
+
+## [v1.5.4](https://github.com/GDay/django-q2/tree/v1.5.4) (2023-06-29)
+
+**Merged pull requests:**
+
+- Rerun successful tasks https://github.com/django-q2/django-q2/pull/99
+
+## [v1.5.3](https://github.com/GDay/django-q2/tree/v1.5.3) (2023-05-14)
+
+**Merged pull requests:**
+
+- Add post_spawn signal. https://github.com/django-q2/django-q2/pull/93
+- Post spawn docs https://github.com/django-q2/django-q2/pull/95
+- Make processes identifiable with uuid4 https://github.com/django-q2/django-q2/pull/91
 
 ## [v1.5.2](https://github.com/GDay/django-q2/tree/v1.5.2) (2023-04-13)
 

README.rst

@@ -27,7 +27,7 @@ Changes compared to the original Django-Q:
 - Dropped support for Disque (hasn't been updated in a long time)
 - Dropped Redis, Arrow and Blessed dependencies
 - Updated all current dependencies
-- Added tests for Django 4.x
+- Added tests for Django 4.x and 5.x
 - Added Turkish language
 - Improved admin area
 - Fixed a lot of issues
@@ -40,7 +40,7 @@ Requirements
 -  `Django <https://www.djangoproject.com>`__ > = 3.2
 -  `Django-picklefield <https://github.com/gintas/django-picklefield>`__
 
-Tested with: Python 3.8, 3.9, 3.10 and 3.11. Works with Django 3.2.X, 4.1.X and 4.2.X.
+Tested with: Python 3.8, 3.9, 3.10, 3.11 and 3.12. Works with Django 3.2.X, 4.1.X, 4.2.X and 5.0.X
 
 Brokers
 ~~~~~~~

django_q/__init__.py

@@ -1,6 +1,6 @@
 import django
 
-VERSION = (1, 5, 2)
+VERSION = (1, 6, 0)
 
 if django.VERSION < (3, 2):
     default_app_config = "django_q.apps.DjangoQConfig"

django_q/admin.py

@@ -10,10 +10,37 @@ from django_q.models import Failure, OrmQ, Schedule, Success, Task
 from django_q.tasks import async_task
 
 
+def resubmit_task(model_admin, request, queryset):
+    """Submit selected tasks back to the queue."""
+    for task in queryset:
+        async_task(
+            task.func,
+            *task.args or (),
+            hook=task.hook,
+            group=task.group,
+            cluster=task.cluster,
+            **task.kwargs or {},
+        )
+        if isinstance(model_admin, FailAdmin):
+            task.delete()
+
+
+resubmit_task.short_description = _("Resubmit selected tasks to queue")
+
+
 class TaskAdmin(admin.ModelAdmin):
     """model admin for success tasks."""
 
-    list_display = ("name", "group", "func", "cluster", "started", "stopped", "time_taken")
+    list_display = (
+        "name",
+        "group",
+        "func",
+        "cluster",
+        "started",
+        "stopped",
+        "time_taken",
+    )
+    actions = [resubmit_task]
 
     def has_add_permission(self, request):
         """Don't allow adds."""
@@ -33,27 +60,24 @@ class TaskAdmin(admin.ModelAdmin):
         return list(self.readonly_fields) + [field.name for field in obj._meta.fields]
 
 
-def retry_failed(FailAdmin, request, queryset):
-    """Submit selected tasks back to the queue."""
-    for task in queryset:
-        async_task(task.func, *task.args or (), hook=task.hook,
-                   group=task.group, cluster=task.cluster, **task.kwargs or {})
-        task.delete()
-
-
-retry_failed.short_description = _("Resubmit selected tasks to queue")
-
-
 class FailAdmin(admin.ModelAdmin):
     """model admin for failed tasks."""
 
-    list_display = ("name", "group", "func", "cluster", "started", "stopped", "short_result")
+    list_display = (
+        "name",
+        "group",
+        "func",
+        "cluster",
+        "started",
+        "stopped",
+        "short_result",
+    )
 
     def has_add_permission(self, request):
         """Don't allow adds."""
         return False
 
-    actions = [retry_failed]
+    actions = [resubmit_task]
     search_fields = ("name", "func", "group")
     list_filter = ("group", "cluster")
     readonly_fields = []
@@ -124,7 +148,17 @@ class QueueAdmin(admin.ModelAdmin):
     """queue admin for ORM broker"""
 
     list_display = ("id", "key", "name", "group", "func", "lock", "task_id")
-    fields = ("key", "lock", "task_id", "name", "group", "func", "args", "kwargs", "q_options")
+    fields = (
+        "key",
+        "lock",
+        "task_id",
+        "name",
+        "group",
+        "func",
+        "args",
+        "kwargs",
+        "q_options",
+    )
     readonly_fields = fields[2:]
 
     def save_model(self, request, obj, form, change):

django_q/brokers/orm.py

@@ -37,7 +37,9 @@ class ORM(Broker):
 
     def lock_size(self) -> int:
         return (
-            self.get_connection().filter(key=self.list_key, lock__gt=timezone.now()).count()
+            self.get_connection()
+            .filter(key=self.list_key, lock__gt=timezone.now())
+            .count()
         )
 
     def purge_queue(self):
@@ -62,7 +64,9 @@ class ORM(Broker):
         return package.pk
 
     def dequeue(self):
-        tasks = self.get_connection().filter(key=self.list_key, lock__lt=timezone.now())[
+        tasks = self.get_connection().filter(
+            key=self.list_key, lock__lt=timezone.now()
+        )[
             0 : Conf.BULK  # noqa: E203
         ]
         if tasks:

django_q/cluster.py

@@ -1,11 +1,7 @@
 # Standard
-import ast
-import pydoc
 import signal
 import socket
-import traceback
 import uuid
-from datetime import datetime, timedelta
 from multiprocessing import Event, Process, Value, current_process
 from time import sleep
 
@@ -24,26 +20,15 @@ from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 
 # Local
-import django_q.tasks
 from django_q.brokers import Broker, get_broker
-from django_q.conf import (
+from django_q.conf import Conf, get_ppid, logger, psutil, setproctitle
-    Conf,
-    croniter,
-    error_reporter,
-    get_ppid,
-    logger,
-    psutil,
-    setproctitle,
-    resource,
-)
 from django_q.humanhash import humanize
-from django_q.models import Schedule, Success, Task
+from django_q.monitor import monitor
+from django_q.pusher import pusher
 from django_q.queues import Queue
-from django_q.signals import post_execute, post_spawn, pre_execute
+from django_q.scheduler import scheduler
-from django_q.signing import BadSignature, SignedPackage
 from django_q.status import Stat, Status
-
+from django_q.worker import worker
-from .utils import get_func_repr, localtime
 
 
 class Cluster:
@@ -69,6 +54,7 @@ class Cluster:
         self.start_event = Event()
         self.sentinel = Process(
             target=Sentinel,
+            name=f"Process-{uuid.uuid4().hex}",
             args=(
                 self.stop_event,
                 self.start_event,
@@ -173,7 +159,7 @@ class Sentinel:
 
     def queue_name(self):
         # multi-queue: cluster name is (broker's) queue_name
-        return self.broker.list_key if self.broker else '--'
+        return self.broker.list_key if self.broker else "--"
 
     def start(self):
         self.broker.ping()
@@ -196,7 +182,7 @@ class Sentinel:
         """
         :type target: function or class
         """
-        p = Process(target=target, args=args)
+        p = Process(target=target, args=args, name=f"Process-{uuid.uuid4().hex}")
         p.daemon = True
         if target == worker:
             p.daemon = Conf.DAEMONIZE_WORKERS
@@ -247,20 +233,22 @@ class Sentinel:
                     try:
                         process_name = psutil.Process(process.pid).name()
                         name_splits = process_name.split(" ")
-                        task_name = name_splits[3] if len(name_splits) >= 4 and name_splits[2] == "processing" else ""
+                        task_name = (
+                            name_splits[3]
+                            if len(name_splits) >= 4 and name_splits[2] == "processing"
+                            else ""
+                        )
                     except psutil.NoSuchProcess:
                         pass
                 process.terminate()
                 if task_name:
-                    msg = (
+                    msg = _(
-                        _("reincarnated worker %(name)s after timeout while processing task %(task_name)s")
+                        "reincarnated worker %(name)s after timeout while processing task %(task_name)s"
-                        % {"name": process.name, "task_name": task_name}
+                    ) % {"name": process.name, "task_name": task_name}
-                    )
                 else:
-                    msg = (
+                    msg = _("reincarnated worker %(name)s after timeout") % {
-                        _("reincarnated worker %(name)s after timeout")
+                        "name": process.name
-                        % {"name": process.name}
+                    }
-                    )
                 logger.critical(msg)
             elif int(process.timer.value) == -2:
                 logger.info(_("recycled worker %(name)s") % {"name": process.name})
@@ -293,14 +281,18 @@ class Sentinel:
             _("%(name)s guarding cluster %(cluster_name)s")
             % {
                 "name": current_process().name,
-                "cluster_name": humanize(self.cluster_id.hex) + f" [{self.queue_name()}]",
+                "cluster_name": humanize(self.cluster_id.hex)
+                + f" [{self.queue_name()}]",
             }
         )
         self.start_event.set()
         Stat(self).save()
         logger.info(
             _("Q Cluster %(cluster_name)s running.")
-            % {"cluster_name": humanize(self.cluster_id.hex) + f" [{self.queue_name()}]"}
+            % {
+                "cluster_name": humanize(self.cluster_id.hex)
+                + f" [{self.queue_name()}]"
+            }
         )
         counter = 0
         cycle = Conf.GUARD_CYCLE  # guard loop sleep in seconds
@@ -373,438 +365,6 @@ class Sentinel:
         Stat(self).save()
 
 
-def pusher(task_queue: Queue, event: Event, broker: Broker = None):
-    """
-    Pulls tasks of the broker and puts them in the task queue
-    :type broker:
-    :type task_queue: multiprocessing.Queue
-    :type event: multiprocessing.Event
-    """
-    if not broker:
-        broker = get_broker()
-    proc_name = current_process().name
-    if setproctitle:
-        setproctitle.setproctitle(f"qcluster {proc_name} pusher")
-    logger.info(
-        _("%(name)s pushing tasks at %(id)s")
-        % {"name": proc_name, "id": current_process().pid}
-    )
-    while True:
-        try:
-            task_set = broker.dequeue()
-        except Exception:
-            logger.exception("Failed to pull task from broker")
-            # broker probably crashed. Let the sentinel handle it.
-            sleep(10)
-            break
-        if task_set:
-            for task in task_set:
-                ack_id = task[0]
-                # unpack the task
-                try:
-                    task = SignedPackage.loads(task[1])
-                except (TypeError, BadSignature):
-                    logger.exception("Failed to push task to queue")
-                    broker.fail(ack_id)
-                    continue
-                task["cluster"] = Conf.CLUSTER_NAME  # save actual cluster name to orm task table
-                task["ack_id"] = ack_id
-                task_queue.put(task)
-            logger.debug(
-                _("queueing from %(list_key)s") % {"list_key": broker.list_key}
-            )
-        if event.is_set():
-            break
-    logger.info(_("%(name)s stopped pushing tasks") % {"name": current_process().name})
-
-
-def monitor(result_queue: Queue, broker: Broker = None):
-    """
-    Gets finished tasks from the result queue and saves them to Django
-    :type broker: brokers.Broker
-    :type result_queue: multiprocessing.Queue
-    """
-    if not broker:
-        broker = get_broker()
-    proc_name = current_process().name
-    if setproctitle:
-        setproctitle.setproctitle(f"qcluster {proc_name} monitor")
-    logger.info(
-        _("%(name)s monitoring at %(id)s") % {"name": proc_name, "id": current_process().pid}
-    )
-    for task in iter(result_queue.get, "STOP"):
-        # save the result
-        if task.get("cached", False):
-            save_cached(task, broker)
-        else:
-            save_task(task, broker)
-        # acknowledge result
-        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
-        info_name = get_func_repr(task["func"])
-        if task["success"]:
-            # log success
-            logger.info(
-                _("Processed '%(info_name)s' (%(task_name)s)")
-                % {"info_name": info_name, "task_name": task["name"]}
-            )
-        else:
-            # log failure
-            logger.error(
-                _("Failed '%(info_name)s' (%(task_name)s) - %(task_result)s")
-                % {
-                    "info_name": info_name,
-                    "task_name": task["name"],
-                    "task_result": task["result"],
-                }
-            )
-    logger.info(_("%(name)s stopped monitoring results") % {"name": proc_name})
-
-
-def worker(
-    task_queue: Queue, result_queue: Queue, timer: Value, timeout: int = Conf.TIMEOUT
-):
-    """
-    Takes a task from the task queue, tries to execute it and puts the result back in
-    the result queue
-    :param timeout: number of seconds wait for a worker to finish.
-    :type task_queue: multiprocessing.Queue
-    :type result_queue: multiprocessing.Queue
-    :type timer: multiprocessing.Value
-    """
-    proc_name = current_process().name
-    logger.info(
-        _("%(proc_name)s ready for work at %(id)s")
-        % {"proc_name": proc_name, "id": current_process().pid}
-    )
-    post_spawn.send(sender="django_q", proc_name=proc_name)
-    if setproctitle:
-        setproctitle.setproctitle(f"qcluster {proc_name} idle")
-    task_count = 0
-    if timeout is None:
-        timeout = -1
-    # Start reading the task queue
-    for task in iter(task_queue.get, "STOP"):
-        result = None
-        timer.value = -1  # Idle
-        task_count += 1
-        f = task["func"]
-
-        # Log task creation and set process name
-        # Get the function from the task
-        func_name = get_func_repr(f)
-        task_name = task["name"]
-        task_desc = (
-            _("%(proc_name)s processing %(task_name)s '%(func_name)s'")
-            % {
-                "proc_name": proc_name,
-                "func_name": func_name,
-                "task_name": task_name,
-            }
-        )
-        if "group" in task:
-            task_desc += f" [{task['group']}]"
-        logger.info(task_desc)
-
-        if setproctitle:
-            proc_title = f"qcluster {proc_name} processing {task_name} '{func_name}'"
-            if "group" in task:
-                proc_title += f" [{task['group']}]"
-            setproctitle.setproctitle(proc_title)
-
-        # if it's not an instance try to get it from the string
-        if not callable(f):
-            # locate() returns None if f cannot be loaded
-            f = pydoc.locate(f)
-        close_old_django_connections()
-        timer_value = task.pop("timeout", timeout)
-        # signal execution
-        pre_execute.send(sender="django_q", func=f, task=task)
-        # execute the payload
-        timer.value = timer_value  # Busy
-
-        try:
-            if f is None:
-                # raise a meaningfull error if task["func"] is not a valid function
-                raise ValueError(f"Function {task['func']} is not defined")
-            res = f(*task["args"], **task["kwargs"])
-            result = (res, True)
-        except Exception as e:
-            result = (f"{e} : {traceback.format_exc()}", False)
-            if error_reporter:
-                error_reporter.report()
-            if task.get("sync", False):
-                raise
-        with timer.get_lock():
-            # Process result
-            task["result"] = result[0]
-            task["success"] = result[1]
-            task["stopped"] = timezone.now()
-            result_queue.put(task)
-            timer.value = -1  # Idle
-            if setproctitle:
-                setproctitle.setproctitle(f"qcluster {proc_name} idle")
-            # Recycle
-            if task_count == Conf.RECYCLE or rss_check():
-                timer.value = -2  # Recycled
-                break
-    logger.info(_("%(proc_name)s stopped doing work") % {"proc_name": proc_name})
-
-
-def save_task(task, broker: Broker):
-    """
-    Saves the task package to Django or the cache
-    :param task: the task package
-    :type broker: brokers.Broker
-    """
-    # SAVE LIMIT < 0 : Don't save success
-    if not task.get("save", Conf.SAVE_LIMIT >= 0) and task["success"]:
-        return
-    # enqueues next in a chain
-    if task.get("chain", None):
-        django_q.tasks.async_chain(
-            task["chain"],
-            group=task["group"],
-            cached=task["cached"],
-            sync=task["sync"],
-            broker=broker,
-        )
-    # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
-    close_old_django_connections()
-
-    try:
-        filters = {}
-        if (
-            Conf.SAVE_LIMIT_PER
-            and Conf.SAVE_LIMIT_PER in {"group", "name", "func"}
-            and Conf.SAVE_LIMIT_PER in task
-        ):
-            value = task[Conf.SAVE_LIMIT_PER]
-            if Conf.SAVE_LIMIT_PER == "func":
-                value = get_func_repr(value)
-            filters[Conf.SAVE_LIMIT_PER] = value
-
-        with db.transaction.atomic(using=db.router.db_for_write(Success)):
-            last = Success.objects.filter(**filters).select_for_update().last()
-            if (
-                task["success"]
-                and 0 < Conf.SAVE_LIMIT <= Success.objects.filter(**filters).count()
-            ):
-                last.delete()
-
-        # check if this task has previous results
-        try:
-            existing_task = Task.objects.get(id=task["id"], name=task["name"])
-            # only update the result if it hasn't succeeded yet
-            if not existing_task.success:
-                existing_task.stopped = task["stopped"]
-                existing_task.result = task["result"]
-                existing_task.success = task["success"]
-                existing_task.attempt_count = existing_task.attempt_count + 1
-                existing_task.save()
-
-            if (
-                Conf.MAX_ATTEMPTS > 0
-                and existing_task.attempt_count >= Conf.MAX_ATTEMPTS
-            ):
-                broker.acknowledge(task["ack_id"])
-
-        except Task.DoesNotExist:
-            # convert func to string
-            func = get_func_repr(task["func"])
-            Task.objects.create(
-                id=task["id"],
-                name=task["name"],
-                func=func,
-                hook=task.get("hook"),
-                args=task["args"],
-                kwargs=task["kwargs"],
-                cluster=task.get("cluster"),
-                started=task["started"],
-                stopped=task["stopped"],
-                result=task["result"],
-                group=task.get("group"),
-                success=task["success"],
-                attempt_count=1,
-            )
-    except Exception:
-        logger.exception("Could not save task result")
-
-
-def save_cached(task, broker: Broker):
-    task_key = f'{broker.list_key}:{task["id"]}'
-    timeout = task["cached"]
-    if timeout is True:
-        timeout = None
-    try:
-        group = task.get("group", None)
-        iter_count = task.get("iter_count", 0)
-        # if it's a group append to the group list
-        if group:
-            group_key = f"{broker.list_key}:{group}:keys"
-            group_list = broker.cache.get(group_key) or []
-            # if it's an iter group, check if we are ready
-            if iter_count and len(group_list) == iter_count - 1:
-                group_args = f"{broker.list_key}:{group}:args"
-                # collate the results into a Task result
-                results = [
-                    SignedPackage.loads(broker.cache.get(k))["result"]
-                    for k in group_list
-                ]
-                results.append(task["result"])
-                task["result"] = results
-                task["id"] = group
-                task["args"] = SignedPackage.loads(broker.cache.get(group_args))
-                task.pop("iter_count", None)
-                task.pop("group", None)
-                if task.get("iter_cached", None):
-                    task["cached"] = task.pop("iter_cached", None)
-                    save_cached(task, broker=broker)
-                else:
-                    save_task(task, broker)
-                broker.cache.delete_many(group_list)
-                broker.cache.delete_many([group_key, group_args])
-                return
-            # save the group list
-            group_list.append(task_key)
-            broker.cache.set(group_key, group_list, timeout)
-            # async_task next in a chain
-            if task.get("chain", None):
-                django_q.tasks.async_chain(
-                    task["chain"],
-                    group=group,
-                    cached=task["cached"],
-                    sync=task["sync"],
-                    broker=broker,
-                )
-        # save the task
-        broker.cache.set(task_key, SignedPackage.dumps(task), timeout)
-    except Exception:
-        logger.exception("Could not save task result")
-
-
-def scheduler(broker: Broker = None):
-    """
-    Creates a task from a schedule at the scheduled time and schedules next run
-    """
-    if not broker:
-        broker = get_broker()
-    close_old_django_connections()
-    try:
-        # Only default cluster will handler schedule with default(null) cluster
-        Q_default = db.models.Q(cluster__isnull=True) if Conf.CLUSTER_NAME == Conf.PREFIX else db.models.Q(pk__in=[])
-
-        with db.transaction.atomic(using=db.router.db_for_write(Schedule)):
-            for s in (
-                Schedule.objects.select_for_update()
-                .exclude(repeats=0)
-                .filter(next_run__lt=timezone.now())
-                .filter(
-                    Q_default | db.models.Q(cluster=Conf.CLUSTER_NAME)
-                )
-            ):
-                args = ()
-                kwargs = {}
-                # get args, kwargs and hook
-                if s.kwargs:
-                    try:
-                        # first try the dict syntax
-                        kwargs = ast.literal_eval(s.kwargs)
-                    except (SyntaxError, ValueError):
-                        # else use the kwargs syntax
-                        try:
-                            parsed_kwargs = (
-                                ast.parse(f"f({s.kwargs})").body[0].value.keywords
-                            )
-                            kwargs = {
-                                kwarg.arg: ast.literal_eval(kwarg.value)
-                                for kwarg in parsed_kwargs
-                            }
-                        except (SyntaxError, ValueError):
-                            kwargs = {}
-                if s.args:
-                    args = ast.literal_eval(s.args)
-                    # single value won't eval to tuple, so:
-                    if type(args) != tuple:
-                        args = (args,)
-                q_options = kwargs.get("q_options", {})
-                if s.intended_date_kwarg:
-                    kwargs[s.intended_date_kwarg] = s.next_run.isoformat()
-                if s.hook:
-                    q_options["hook"] = s.hook
-                # set up the next run time
-                if s.schedule_type != s.ONCE:
-                    next_run = s.next_run
-                    while True:
-                        next_run = s.calculate_next_run(next_run)
-                        if Conf.CATCH_UP or next_run > localtime():
-                            break
-
-                    s.next_run = next_run
-                    s.repeats += -1
-                # send it to the cluster; any cluster name is allowed in multi-queue scenarios
-                # because `broker_name` is confusing, using `cluster` name is recommended and takes precedence
-                q_options["cluster"] = s.cluster or q_options.get("cluster", q_options.pop("broker_name", None))
-                if q_options['cluster'] is None or q_options['cluster'] == Conf.CLUSTER_NAME:
-                    q_options["broker"] = broker
-                q_options["group"] = q_options.get("group", s.name or s.id)
-                kwargs["q_options"] = q_options
-                s.task = django_q.tasks.async_task(s.func, *args, **kwargs)
-                # log it
-                if not s.task:
-                    logger.error(
-                        _(
-                            "%(process_name)s failed to create a task from schedule "
-                            "[%(schedule)s]"
-                        )
-                        % {
-                            "process_name": current_process().name,
-                            "schedule": s.name or s.id,
-                        }
-                    )
-                else:
-                    logger.info(
-                        _(
-                            "%(process_name)s created task %(task_name)s from schedule "
-                            "[%(schedule)s]"
-                        )
-                        % {
-                            "process_name": current_process().name,
-                            "task_name": humanize(s.task),
-                            "schedule": s.name or s.id,
-                        }
-                    )
-                # default behavior is to delete a ONCE schedule
-                if s.schedule_type == s.ONCE:
-                    if s.repeats < 0:
-                        s.delete()
-                        continue
-                    # but not if it has a positive repeats
-                    s.repeats = 0
-                # save the schedule
-                s.save()
-    except Exception:
-        logger.exception("Could not create task from schedule")
-
-
-def close_old_django_connections():
-    """
-    Close django connections unless running with sync=True.
-    """
-    if Conf.SYNC:
-        logger.warning(
-            "Preserving django database connections because sync=True. Beware "
-            "that tasks are now injected in the calling context/transactions "
-            "which may result in unexpected behaviour."
-        )
-    else:
-        db.close_old_connections()
-
-
 def set_cpu_affinity(n: int, process_ids: list, actual: bool = not Conf.TESTING):
     """
     Sets the cpu affinity for the supplied processes.
@@ -845,12 +405,3 @@ def set_cpu_affinity(n: int, process_ids: list, actual: bool = not Conf.TESTING)
                 _("%(pid)s will use cpu %(affinity)s")
                 % {"pid": pid, "affinity": affinity}
             )
-
-
-def rss_check():
-    if Conf.MAX_RSS:
-        if resource:
-            return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss >= Conf.MAX_RSS
-        elif psutil:
-            return psutil.Process().memory_info().rss >= Conf.MAX_RSS * 1024
-    return False

django_q/conf.py

@@ -1,16 +1,22 @@
 import logging
 import os
+import sys
 from copy import deepcopy
 from multiprocessing import cpu_count
 from signal import signal
 from warnings import warn
 
-import pkg_resources
 from django.conf import settings
 from django.utils.translation import gettext_lazy as _
 
 from django_q.queues import Queue
 
+# The "selectable" entry points were introduced in importlib_metadata 3.6 and Python 3.10.
+if sys.version_info < (3, 10):
+    from importlib_metadata import entry_points
+else:
+    from importlib.metadata import entry_points
+
 # optional
 try:
     import psutil
@@ -44,15 +50,18 @@ class Conf:
         conf = {}
 
     _Q_CLUSTER_NAME = os.getenv("Q_CLUSTER_NAME")
-    if _Q_CLUSTER_NAME and _Q_CLUSTER_NAME != conf.get("name") and \
+    if (
-            _Q_CLUSTER_NAME != conf.get("cluster_name"):
+        _Q_CLUSTER_NAME
+        and _Q_CLUSTER_NAME != conf.get("name")
+        and _Q_CLUSTER_NAME != conf.get("cluster_name")
+    ):
         conf["cluster_name"] = _Q_CLUSTER_NAME
         alt_conf = conf.pop("ALT_CLUSTERS")
         if isinstance(alt_conf, dict):
             alt_conf = alt_conf.get(_Q_CLUSTER_NAME)
             if isinstance(alt_conf, dict):
-                alt_conf.pop('name', None)
+                alt_conf.pop("name", None)
-                alt_conf.pop('cluster_name', None)
+                alt_conf.pop("cluster_name", None)
                 conf.update(alt_conf)
 
     # Redis server configuration . Follows standard redis keywords
@@ -273,9 +282,7 @@ if Conf.ERROR_REPORTER:
         # iterate through the configured error reporters,
         # and instantiate an ErrorReporter using the provided config
         for name, conf in error_conf.items():
-            for entry in pkg_resources.iter_entry_points(
+            for entry in entry_points(group="djangoq.errorreporters", name=name):
-                "djangoq.errorreporters", name
-            ):
                 Reporter = entry.load()
                 reporters.append(Reporter(**conf))
         error_reporter = ErrorReporter(reporters)

django_q/core_signing.py

@@ -2,16 +2,18 @@ import datetime
 import time
 import zlib
 
+import django
 from django.core.signing import BadSignature, JSONSerializer, SignatureExpired
 from django.core.signing import Signer as Sgnr
 from django.core.signing import TimestampSigner as TsS
 from django.core.signing import b64_decode, dumps
 
-try:
+if django.VERSION < (5, 0):
-    from django.core.signing import base62
-except ImportError:
-    # For django < 4.0
     from django.utils.baseconv import base62
+    b62_decode = base62.decode
+else:
+    from django.core.signing import b62_decode
+
 from django.utils.crypto import constant_time_compare
 from django.utils.encoding import force_bytes, force_str
 
@@ -37,7 +39,9 @@ def loads(
     """
     # TimestampSigner.unsign() returns str but base64 and zlib compression
     # operate on bytes.
-    base64d = force_bytes(TimestampSigner(key=key, salt=salt).unsign(s, max_age=max_age))
+    base64d = force_bytes(
+        TimestampSigner(key=key, salt=salt).unsign(s, max_age=max_age)
+    )
     decompress = False
     if base64d[:1] == b".":
         # It's compressed; uncompress it first
@@ -74,7 +78,7 @@ class TimestampSigner(Signer, TsS):
         """
         result = super(TimestampSigner, self).unsign(value)
         value, timestamp = result.rsplit(self.sep, 1)
-        timestamp = base62.decode(timestamp)
+        timestamp = b62_decode(timestamp)
         if max_age is not None:
             if isinstance(max_age, datetime.timedelta):
                 max_age = max_age.total_seconds()

django_q/locale/de/LC_MESSAGES/django.po

@@ -318,7 +318,7 @@ msgstr "Anzahl Minuten für den Typ 'Minuten'"
 
 #: models.py:205
 msgid "Repeats"
-msgstr "Wiederhohlungen"
+msgstr "Wiederholungen"
 
 #: models.py:205
 msgid "n = n times, -1 = forever"

django_q/management/commands/qcluster.py

@@ -1,8 +1,9 @@
+import os
+
 from django.core.management.base import BaseCommand
 from django.utils.translation import gettext as _
 
 from django_q.cluster import Cluster
-import os
 
 
 class Command(BaseCommand):
@@ -23,7 +24,7 @@ class Command(BaseCommand):
             dest="cluster_name",
             default=None,
             help="Set alternative cluster name instead of the name in Q_CLUSTER settings (for multi-queue setup). "
-                 "On Linux you should set name through `Q_CLUSTER_NAME=cluster_name python manage.py qcluster` instead."
+            "On Linux you should set name through `Q_CLUSTER_NAME=cluster_name python manage.py qcluster` instead.",
         )
 
     def handle(self, *args, **options):

django_q/management/commands/qinfo.py

@@ -3,7 +3,7 @@ from django.utils.translation import gettext as _
 
 from django_q import VERSION
 from django_q.conf import Conf
-from django_q.monitor import get_ids, info
+from django_q.monitor_terminal import get_ids, info
 
 
 class Command(BaseCommand):

django_q/management/commands/qmemory.py

@@ -1,7 +1,7 @@
 from django.core.management.base import BaseCommand
 from django.utils.translation import gettext as _
 
-from django_q.monitor import memory
+from django_q.monitor_terminal import memory
 
 
 class Command(BaseCommand):

django_q/management/commands/qmonitor.py

@@ -1,7 +1,7 @@
 from django.core.management.base import BaseCommand
 from django.utils.translation import gettext as _
 
-from django_q.monitor import monitor
+from django_q.monitor_terminal import monitor
 
 
 class Command(BaseCommand):

django_q/migrations/0016_schedule_intended_date_kwarg.py

@@ -1,6 +1,7 @@
 # Generated by Django 4.1.2 on 2023-01-15 22:34
 
 from django.db import migrations, models
+
 import django_q.models
 
 

django_q/models.py

@@ -8,19 +8,19 @@ from django.db import models
 from django.template.defaultfilters import truncatechars
 from django.urls import reverse
 from django.utils import timezone
-from django.utils.timezone import is_aware
-from django.utils.html import format_html
-from django.utils.translation import gettext_lazy as _
 from django.utils.functional import cached_property
+from django.utils.html import format_html
+from django.utils.timezone import is_aware
+from django.utils.translation import gettext_lazy as _
 
 # External
 from picklefield import PickledObjectField
 from picklefield.fields import dbsafe_decode
 
 # Local
-from django_q.conf import croniter, Conf
+from django_q.conf import Conf, croniter
 from django_q.signing import SignedPackage
-from django_q.utils import localtime, add_months, add_years
+from django_q.utils import add_months, add_years, localtime
 
 from .utils import get_func_repr
 
@@ -218,8 +218,11 @@ class Schedule(models.Model):
     )
     task = models.CharField(max_length=100, null=True, editable=False)
     cluster = models.CharField(
-        max_length=100, default=None, null=True, blank=True,
+        max_length=100,
-        help_text=_("Name of the target cluster")
+        default=None,
+        null=True,
+        blank=True,
+        help_text=_("Name of the target cluster"),
     )
     intended_date_kwarg = models.CharField(
         max_length=100,
@@ -309,7 +312,9 @@ class Schedule(models.Model):
 class OrmQ(models.Model):
     key = models.CharField(max_length=100, help_text=_("Name of the target cluster"))
     payload = models.TextField()
-    lock = models.DateTimeField(null=True, help_text=_("Prevent any cluster from pulling until"))
+    lock = models.DateTimeField(
+        null=True, help_text=_("Prevent any cluster from pulling until")
+    )
 
     @cached_property
     def task(self):

django_q/monitor.py

@@ -1,510 +1,206 @@
-from datetime import timedelta
+from multiprocessing.process import current_process
+from multiprocessing.queues import Queue
 
-# django
+from django import core, db
-from django.db import connection
+from django.utils.translation import gettext_lazy as _
-from django.db.models import F, Sum
+from django.apps.registry import apps
-from django.utils import timezone
-from django.utils.translation import gettext as _
 
-from django_q import VERSION, models
-from django_q.brokers import get_broker
-
-# local
-from django_q.conf import Conf
-from django_q.status import Stat
-
-# optional
 try:
-    import psutil
+    apps.check_apps_ready()
-except ImportError:
+except core.exceptions.AppRegistryNotReady:
-    psutil = None
+    import django
+
+    django.setup()
+
+from django_q.brokers import Broker, get_broker
+from django_q.conf import Conf, logger, setproctitle
+from django_q.models import Success, Task
+from django_q.signals import post_execute
+from django_q.signing import SignedPackage
+from django_q.tasks import async_chain
+from django_q.utils import close_old_django_connections, get_func_repr
+
+try:
+    import setproctitle
+except ModuleNotFoundError:
+    setproctitle = None
 
 
-def get_process_mb(pid):
+def monitor(result_queue: Queue, broker: Broker = None):
-    try:
+    """
-        process = psutil.Process(pid)
+    Gets finished tasks from the result queue and saves them to Django
-        mb_used = round(process.memory_info().rss / 1024**2, 2)
+    :type broker: brokers.Broker
-    except psutil.NoSuchProcess:
+    :type result_queue: multiprocessing.Queue
-        mb_used = "NO_PROCESS_FOUND"
+    """
-    return mb_used
-
-
-BLESSED_INSTALL_MESSAGE = (
-    "Blessed is not installed. Please install blessed to use this: "
-    "https://pypi.org/project/blessed/"
-)
-
-
-def monitor(run_once=False, broker=None):
     if not broker:
         broker = get_broker()
-    try:
+    proc_name = current_process().name
-        from blessed import Terminal
+    if setproctitle:
+        setproctitle.setproctitle(f"qcluster {proc_name} monitor")
+    logger.info(
+        _("%(name)s monitoring at %(id)s")
+        % {"name": proc_name, "id": current_process().pid}
+    )
+    for task in iter(result_queue.get, "STOP"):
+        # save the result
+        if task.get("cached", False):
+            save_cached(task, broker)
+        else:
+            save_task(task, broker)
+        # acknowledge result
+        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
+        info_name = get_func_repr(task["func"])
+        if task["success"]:
+            # log success
+            logger.info(
+                _("Processed '%(info_name)s' (%(task_name)s)")
+                % {"info_name": info_name, "task_name": task["name"]}
+            )
+        else:
+            # log failure
+            logger.error(
+                _("Failed '%(info_name)s' (%(task_name)s) - %(task_result)s")
+                % {
+                    "info_name": info_name,
+                    "task_name": task["name"],
+                    "task_result": task["result"],
+                }
+            )
+    logger.info(_("%(name)s stopped monitoring results") % {"name": proc_name})
 
-        term = Terminal()
+
-    except ImportError:
+def save_task(task, broker: Broker):
-        print(BLESSED_INSTALL_MESSAGE)
+    """
+    Saves the task package to Django or the cache
+    :param task: the task package
+    :type broker: brokers.Broker
+    """
+    # SAVE LIMIT < 0 : Don't save success
+    if not task.get("save", Conf.SAVE_LIMIT >= 0) and task["success"]:
         return
+    # enqueues next in a chain
+    if task.get("chain", None):
+        async_chain(
+            task["chain"],
+            group=task["group"],
+            cached=task["cached"],
+            sync=task["sync"],
+            broker=broker,
+        )
+    # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
+    close_old_django_connections()
 
-    broker.ping()
+    try:
-    with term.fullscreen(), term.hidden_cursor(), term.cbreak():
+        filters = {}
-        val = None
+        if (
-        start_width = int(term.width / 8)
+            Conf.SAVE_LIMIT_PER
-        while val not in (
+            and Conf.SAVE_LIMIT_PER in {"group", "name", "func"}
-            "q",
+            and Conf.SAVE_LIMIT_PER in task
-            "Q",
         ):
-            col_width = int(term.width / 8)
+            value = task[Conf.SAVE_LIMIT_PER]
-            # In case of resize
+            if Conf.SAVE_LIMIT_PER == "func":
-            if col_width != start_width:
+                value = get_func_repr(value)
-                print(term.clear())
+            filters[Conf.SAVE_LIMIT_PER] = value
-                start_width = col_width
+
-            print(
+        with db.transaction.atomic(using=db.router.db_for_write(Success)):
-                term.move(0, 0)
+            last = Success.objects.filter(**filters).select_for_update().last()
-                + term.black_on_green(term.center(_("Host"), width=col_width - 1))
+            if (
+                task["success"]
+                and 0 < Conf.SAVE_LIMIT <= Success.objects.filter(**filters).count()
+            ):
+                last.delete()
+
+        # check if this task has previous results
+        try:
+            existing_task = Task.objects.get(id=task["id"], name=task["name"])
+            # only update the result if it hasn't succeeded yet
+            if not existing_task.success:
+                existing_task.stopped = task["stopped"]
+                existing_task.result = task["result"]
+                existing_task.success = task["success"]
+                existing_task.attempt_count = existing_task.attempt_count + 1
+                existing_task.save()
+
+            if (
+                Conf.MAX_ATTEMPTS > 0
+                and existing_task.attempt_count >= Conf.MAX_ATTEMPTS
+            ):
+                broker.acknowledge(task["ack_id"])
+
+        except Task.DoesNotExist:
+            # convert func to string
+            func = get_func_repr(task["func"])
+            Task.objects.create(
+                id=task["id"],
+                name=task["name"],
+                func=func,
+                hook=task.get("hook"),
+                args=task["args"],
+                kwargs=task["kwargs"],
+                cluster=task.get("cluster"),
+                started=task["started"],
+                stopped=task["stopped"],
+                result=task["result"],
+                group=task.get("group"),
+                success=task["success"],
+                attempt_count=1,
             )
-            print(
+    except Exception:
-                term.move(0, 1 * col_width)
+        logger.exception("Could not save task result")
-                + term.black_on_green(term.center(_("Id"), width=col_width - 1))
-            )
-            print(
-                term.move(0, 2 * col_width)
-                + term.black_on_green(term.center(_("State"), width=col_width - 1))
-            )
-            print(
-                term.move(0, 3 * col_width)
-                + term.black_on_green(term.center(_("Pool"), width=col_width - 1))
-            )
-            print(
-                term.move(0, 4 * col_width)
-                + term.black_on_green(term.center(_("TQ"), width=col_width - 1))
-            )
-            print(
-                term.move(0, 5 * col_width)
-                + term.black_on_green(term.center(_("RQ"), width=col_width - 1))
-            )
-            print(
-                term.move(0, 6 * col_width)
-                + term.black_on_green(term.center(_("RC"), width=col_width - 1))
-            )
-            print(
-                term.move(0, 7 * col_width)
-                + term.black_on_green(term.center(_("Up"), width=col_width - 1))
-            )
-            i = 2
-            stats = Stat.get_all(broker=broker)
-            print(term.clear_eos())
-            for stat in stats:
-                status = stat.status
-                # color status
-                if stat.status == Conf.WORKING:
-                    status = term.green(str(Conf.WORKING))
-                elif stat.status == Conf.STOPPING:
-                    status = term.yellow(str(Conf.STOPPING))
-                elif stat.status == Conf.STOPPED:
-                    status = term.red(str(Conf.STOPPED))
-                elif stat.status == Conf.IDLE:
-                    status = str(Conf.IDLE)
-                # color q's
-                tasks = str(stat.task_q_size)
-                if stat.task_q_size > 0:
-                    tasks = term.cyan(str(stat.task_q_size))
-                    if Conf.QUEUE_LIMIT and stat.task_q_size == Conf.QUEUE_LIMIT:
-                        tasks = term.green(str(stat.task_q_size))
-                results = stat.done_q_size
-                if results > 0:
-                    results = term.cyan(str(results))
-                # color workers
-                workers = len(stat.workers)
-                if workers < Conf.WORKERS:
-                    workers = term.yellow(str(workers))
-                # format uptime
-                uptime = (timezone.now() - stat.tob).total_seconds()
-                hours, remainder = divmod(uptime, 3600)
-                minutes, seconds = divmod(remainder, 60)
-                uptime = "%d:%02d:%02d" % (hours, minutes, seconds)
-                # print to the terminal
-                print(
-                    term.move(i, 0)
-                    + term.center(stat.host[: col_width - 1], width=col_width - 1)
-                )
-                print(
-                    term.move(i, 1 * col_width)
-                    + term.center(str(stat.cluster_id)[-8:], width=col_width - 1)
-                )
-                print(
-                    term.move(i, 2 * col_width)
-                    + term.center(status, width=col_width - 1)
-                )
-                print(
-                    term.move(i, 3 * col_width)
-                    + term.center(workers, width=col_width - 1)
-                )
-                print(
-                    term.move(i, 4 * col_width)
-                    + term.center(tasks, width=col_width - 1)
-                )
-                print(
-                    term.move(i, 5 * col_width)
-                    + term.center(results, width=col_width - 1)
-                )
-                print(
-                    term.move(i, 6 * col_width)
-                    + term.center(stat.reincarnations, width=col_width - 1)
-                )
-                print(
-                    term.move(i, 7 * col_width)
-                    + term.center(uptime, width=col_width - 1)
-                )
-                i += 1
-            # bottom bar
-            i += 1
-            queue_size = broker.queue_size()
-            lock_size = broker.lock_size()
-            if lock_size:
-                queue_size = f"{queue_size}({lock_size})"
-            print(
-                term.move(i, 0)
-                + term.white_on_cyan(term.center(broker.info(), width=col_width * 2))
-            )
-            print(
-                term.move(i, 2 * col_width)
-                + term.black_on_cyan(term.center(_("Queued"), width=col_width))
-            )
-            print(
-                term.move(i, 3 * col_width)
-                + term.white_on_cyan(term.center(queue_size, width=col_width))
-            )
-            print(
-                term.move(i, 4 * col_width)
-                + term.black_on_cyan(term.center(_("Success"), width=col_width))
-            )
-            print(
-                term.move(i, 5 * col_width)
-                + term.white_on_cyan(
-                    term.center(models.Success.objects.count(), width=col_width)
-                )
-            )
-            print(
-                term.move(i, 6 * col_width)
-                + term.black_on_cyan(term.center(_("Failures"), width=col_width))
-            )
-            print(
-                term.move(i, 7 * col_width)
-                + term.white_on_cyan(
-                    term.center(models.Failure.objects.count(), width=col_width)
-                )
-            )
-            # for testing
-            if run_once:
-                return Stat.get_all(broker=broker)
-            print(term.move(i + 2, 0) + term.center(_("[Press q to quit]")))
-            val = term.inkey(timeout=1)
 
 
-def info(broker=None):
+def save_cached(task, broker: Broker):
-    if not broker:
+    task_key = f'{broker.list_key}:{task["id"]}'
-        broker = get_broker()
+    timeout = task["cached"]
+    if timeout is True:
+        timeout = None
     try:
-        from blessed import Terminal
+        group = task.get("group", None)
-
+        iter_count = task.get("iter_count", 0)
-        term = Terminal()
+        # if it's a group append to the group list
-    except ImportError:
+        if group:
-        print(BLESSED_INSTALL_MESSAGE)
+            group_key = f"{broker.list_key}:{group}:keys"
-        return
+            group_list = broker.cache.get(group_key) or []
-
+            # if it's an iter group, check if we are ready
-    broker.ping()
+            if iter_count and len(group_list) == iter_count - 1:
-    stat = Stat.get_all(broker=broker)
+                group_args = f"{broker.list_key}:{group}:args"
-    # general stats
+                # collate the results into a Task result
-    clusters = len(stat)
+                results = [
-    workers = 0
+                    SignedPackage.loads(broker.cache.get(k))["result"]
-    reincarnations = 0
+                    for k in group_list
-    for cluster in stat:
+                ]
-        workers += len(cluster.workers)
+                results.append(task["result"])
-        reincarnations += cluster.reincarnations
+                task["result"] = results
-    # calculate tasks pm and avg exec time
+                task["id"] = group
-    tasks_per = 0
+                task["args"] = SignedPackage.loads(broker.cache.get(group_args))
-    per = _("day")
+                task.pop("iter_count", None)
-    exec_time = 0
+                task.pop("group", None)
-    last_tasks = models.Success.objects.filter(
+                if task.get("iter_cached", None):
-        stopped__gte=timezone.now() - timedelta(hours=24)
+                    task["cached"] = task.pop("iter_cached", None)
-    )
+                    save_cached(task, broker=broker)
-    tasks_per_day = last_tasks.count()
+                else:
-    if tasks_per_day > 0:
+                    save_task(task, broker)
-        # average execution time over the last 24 hours
+                broker.cache.delete_many(group_list)
-        if connection.vendor != "sqlite":
+                broker.cache.delete_many([group_key, group_args])
-            exec_time = last_tasks.aggregate(
+                return
-                time_taken=Sum(F("stopped") - F("started"))
+            # save the group list
-            )
+            group_list.append(task_key)
-            exec_time = exec_time["time_taken"].total_seconds() / tasks_per_day
+            broker.cache.set(group_key, group_list, timeout)
-        else:
+            # async_task next in a chain
-            # can't sum timedeltas on sqlite
+            if task.get("chain", None):
-            for t in last_tasks:
+                async_chain(
-                exec_time += t.time_taken()
+                    task["chain"],
-            exec_time = exec_time / tasks_per_day
+                    group=group,
-        # tasks per second/minute/hour/day in the last 24 hours
+                    cached=task["cached"],
-        if tasks_per_day > 24 * 60 * 60:
+                    sync=task["sync"],
-            tasks_per = tasks_per_day / (24 * 60 * 60)
+                    broker=broker,
-            per = _("second")
-        elif tasks_per_day > 24 * 60:
-            tasks_per = tasks_per_day / (24 * 60)
-            per = _("minute")
-        elif tasks_per_day > 24:
-            tasks_per = tasks_per_day / 24
-            per = _("hour")
-        else:
-            tasks_per = tasks_per_day
-    # print to terminal
-    print(term.clear_eos())
-    col_width = int(term.width / 6)
-    print(
-        term.black_on_green(
-            term.center(
-                _("-- %(prefix)s %(version)s on %(info)s --")
-                % {
-                    "prefix": Conf.PREFIX.capitalize(),
-                    "version": ".".join(str(v) for v in VERSION),
-                    "info": broker.info(),
-                }
-            )
-        )
-    )
-    print(
-        term.cyan(_("Clusters"))
-        + term.move_x(1 * col_width)
-        + term.white(str(clusters))
-        + term.move_x(2 * col_width)
-        + term.cyan(_("Workers"))
-        + term.move_x(3 * col_width)
-        + term.white(str(workers))
-        + term.move_x(4 * col_width)
-        + term.cyan(_("Restarts"))
-        + term.move_x(5 * col_width)
-        + term.white(str(reincarnations))
-    )
-    print(
-        term.cyan(_("Queued"))
-        + term.move_x(1 * col_width)
-        + term.white(str(broker.queue_size()))
-        + term.move_x(2 * col_width)
-        + term.cyan(_("Successes"))
-        + term.move_x(3 * col_width)
-        + term.white(str(models.Success.objects.count()))
-        + term.move_x(4 * col_width)
-        + term.cyan(_("Failures"))
-        + term.move_x(5 * col_width)
-        + term.white(str(models.Failure.objects.count()))
-    )
-    print(
-        term.cyan(_("Schedules"))
-        + term.move_x(1 * col_width)
-        + term.white(str(models.Schedule.objects.count()))
-        + term.move_x(2 * col_width)
-        + term.cyan(_("Tasks/%(per)s") % {"per": per})
-        + term.move_x(3 * col_width)
-        + term.white(f"{tasks_per:.2f}")
-        + term.move_x(4 * col_width)
-        + term.cyan(_("Avg time"))
-        + term.move_x(5 * col_width)
-        + term.white(f"{exec_time:.4f}")
-    )
-    return True
-
-
-def memory(run_once=False, workers=False, broker=None):
-    if not broker:
-        broker = get_broker()
-    try:
-        from blessed import Terminal
-
-        term = Terminal()
-    except ImportError:
-        print(BLESSED_INSTALL_MESSAGE)
-        return
-    broker.ping()
-    if not psutil:
-        print(term.clear_eos())
-        print(
-            term.white_on_red(
-                'Cannot start "qmemory" command. Missing "psutil" library.'
-            )
-        )
-        return
-    with term.fullscreen(), term.hidden_cursor(), term.cbreak():
-        MEMORY_AVAILABLE_LOWEST_PERCENTAGE = 100.0
-        MEMORY_AVAILABLE_LOWEST_PERCENTAGE_AT = timezone.now()
-        cols = 8
-        val = None
-        start_width = int(term.width / cols)
-        while val not in ["q", "Q"]:
-            col_width = int(term.width / cols)
-            # In case of resize
-            if col_width != start_width:
-                print(term.clear())
-                start_width = col_width
-            # sentinel, monitor and workers memory usage
-            print(
-                term.move(0, 0 * col_width)
-                + term.black_on_green(term.center(_("Host"), width=col_width - 1))
-            )
-            print(
-                term.move(0, 1 * col_width)
-                + term.black_on_green(term.center(_("Id"), width=col_width - 1))
-            )
-            print(
-                term.move(0, 2 * col_width)
-                + term.black_on_green(
-                    term.center(_("Available (%)"), width=col_width - 1)
                 )
-            )
+        # save the task
-            print(
+        broker.cache.set(task_key, SignedPackage.dumps(task), timeout)
-                term.move(0, 3 * col_width)
+    except Exception:
-                + term.black_on_green(
+        logger.exception("Could not save task result")
-                    term.center(_("Available (MB)"), width=col_width - 1)
-                )
-            )
-            print(
-                term.move(0, 4 * col_width)
-                + term.black_on_green(term.center(_("Total (MB)"), width=col_width - 1))
-            )
-            print(
-                term.move(0, 5 * col_width)
-                + term.black_on_green(
-                    term.center(_("Sentinel (MB)"), width=col_width - 1)
-                )
-            )
-            print(
-                term.move(0, 6 * col_width)
-                + term.black_on_green(
-                    term.center(_("Monitor (MB)"), width=col_width - 1)
-                )
-            )
-            print(
-                term.move(0, 7 * col_width)
-                + term.black_on_green(
-                    term.center(_("Workers (MB)"), width=col_width - 1)
-                )
-            )
-            row = 2
-            stats = Stat.get_all(broker=broker)
-            print(term.clear_eos())
-            for stat in stats:
-                # memory available (%)
-                memory_available_percentage = round(
-                    psutil.virtual_memory().available
-                    * 100
-                    / psutil.virtual_memory().total,
-                    2,
-                )
-                # memory available (MB)
-                memory_available = round(
-                    psutil.virtual_memory().available / 1024**2, 2
-                )
-                if memory_available_percentage < MEMORY_AVAILABLE_LOWEST_PERCENTAGE:
-                    MEMORY_AVAILABLE_LOWEST_PERCENTAGE = memory_available_percentage
-                    MEMORY_AVAILABLE_LOWEST_PERCENTAGE_AT = timezone.now()
-                print(
-                    term.move(row, 0 * col_width)
-                    + term.center(stat.host[: col_width - 1], width=col_width - 1)
-                )
-                print(
-                    term.move(row, 1 * col_width)
-                    + term.center(str(stat.cluster_id)[-8:], width=col_width - 1)
-                )
-                print(
-                    term.move(row, 2 * col_width)
-                    + term.center(memory_available_percentage, width=col_width - 1)
-                )
-                print(
-                    term.move(row, 3 * col_width)
-                    + term.center(memory_available, width=col_width - 1)
-                )
-                print(
-                    term.move(row, 4 * col_width)
-                    + term.center(
-                        round(psutil.virtual_memory().total / 1024**2, 2),
-                        width=col_width - 1,
-                    )
-                )
-                print(
-                    term.move(row, 5 * col_width)
-                    + term.center(get_process_mb(stat.sentinel), width=col_width - 1)
-                )
-                print(
-                    term.move(row, 6 * col_width)
-                    + term.center(
-                        get_process_mb(getattr(stat, "monitor", None)),
-                        width=col_width - 1,
-                    )
-                )
-                workers_mb = 0
-                for worker_pid in stat.workers:
-                    result = get_process_mb(worker_pid)
-                    if isinstance(result, str):
-                        result = 0
-                    workers_mb += result
-                print(
-                    term.move(row, 7 * col_width)
-                    + term.center(
-                        workers_mb or "NO_PROCESSES_FOUND", width=col_width - 1
-                    )
-                )
-                row += 1
-            # each worker's memory usage
-            if workers:
-                row += 2
-                col_width = int(term.width / (1 + Conf.WORKERS))
-                print(
-                    term.move(row, 0 * col_width)
-                    + term.black_on_cyan(term.center(_("Id"), width=col_width - 1))
-                )
-                for worker_num in range(Conf.WORKERS):
-                    print(
-                        term.move(row, (worker_num + 1) * col_width)
-                        + term.black_on_cyan(
-                            term.center(
-                                "Worker #{} (MB)".format(worker_num + 1),
-                                width=col_width - 1,
-                            )
-                        )
-                    )
-                row += 2
-                for stat in stats:
-                    print(
-                        term.move(row, 0 * col_width)
-                        + term.center(str(stat.cluster_id)[-8:], width=col_width - 1)
-                    )
-                    for idx, worker_pid in enumerate(stat.workers):
-                        mb_used = get_process_mb(worker_pid)
-                        print(
-                            term.move(row, (idx + 1) * col_width)
-                            + term.center(mb_used, width=col_width - 1)
-                        )
-                    row += 1
-            row += 1
-            print(
-                term.move(row, 0)
-                + _("Available lowest (): %(memory_percent)s ((at)s)")
-                % {
-                    "memory_percent": str(MEMORY_AVAILABLE_LOWEST_PERCENTAGE),
-                    "at": MEMORY_AVAILABLE_LOWEST_PERCENTAGE_AT.strftime(
-                        "%Y-%m-%d %H:%M:%S+00:00"
-                    ),
-                }
-            )
-            # for testing
-            if run_once:
-                return Stat.get_all(broker=broker)
-            print(term.move(row + 2, 0) + term.center(_("[Press q to quit]")))
-            val = term.inkey(timeout=1)
-
-
-def get_ids():
-    # prints id (PID) of running clusters
-    stat = Stat.get_all()
-    if stat:
-        for s in stat:
-            print(s.cluster_id)
-    else:
-        print(_("No clusters appear to be running."))
-    return True

django_q/monitor_terminal.py

@@ -0,0 +1,510 @@
+from datetime import timedelta
+
+# django
+from django.db import connection
+from django.db.models import F, Sum
+from django.utils import timezone
+from django.utils.translation import gettext as _
+
+from django_q import VERSION, models
+from django_q.brokers import get_broker
+
+# local
+from django_q.conf import Conf
+from django_q.status import Stat
+
+# optional
+try:
+    import psutil
+except ImportError:
+    psutil = None
+
+
+def get_process_mb(pid):
+    try:
+        process = psutil.Process(pid)
+        mb_used = round(process.memory_info().rss / 1024**2, 2)
+    except psutil.NoSuchProcess:
+        mb_used = "NO_PROCESS_FOUND"
+    return mb_used
+
+
+BLESSED_INSTALL_MESSAGE = (
+    "Blessed is not installed. Please install blessed to use this: "
+    "https://pypi.org/project/blessed/"
+)
+
+
+def monitor(run_once=False, broker=None):
+    if not broker:
+        broker = get_broker()
+    try:
+        from blessed import Terminal
+
+        term = Terminal()
+    except ImportError:
+        print(BLESSED_INSTALL_MESSAGE)
+        return
+
+    broker.ping()
+    with term.fullscreen(), term.hidden_cursor(), term.cbreak():
+        val = None
+        start_width = int(term.width / 8)
+        while val not in (
+            "q",
+            "Q",
+        ):
+            col_width = int(term.width / 8)
+            # In case of resize
+            if col_width != start_width:
+                print(term.clear())
+                start_width = col_width
+            print(
+                term.move(0, 0)
+                + term.black_on_green(term.center(_("Host"), width=col_width - 1))
+            )
+            print(
+                term.move(0, 1 * col_width)
+                + term.black_on_green(term.center(_("Id"), width=col_width - 1))
+            )
+            print(
+                term.move(0, 2 * col_width)
+                + term.black_on_green(term.center(_("State"), width=col_width - 1))
+            )
+            print(
+                term.move(0, 3 * col_width)
+                + term.black_on_green(term.center(_("Pool"), width=col_width - 1))
+            )
+            print(
+                term.move(0, 4 * col_width)
+                + term.black_on_green(term.center(_("TQ"), width=col_width - 1))
+            )
+            print(
+                term.move(0, 5 * col_width)
+                + term.black_on_green(term.center(_("RQ"), width=col_width - 1))
+            )
+            print(
+                term.move(0, 6 * col_width)
+                + term.black_on_green(term.center(_("RC"), width=col_width - 1))
+            )
+            print(
+                term.move(0, 7 * col_width)
+                + term.black_on_green(term.center(_("Up"), width=col_width - 1))
+            )
+            i = 2
+            stats = Stat.get_all(broker=broker)
+            print(term.clear_eos())
+            for stat in stats:
+                status = stat.status
+                # color status
+                if stat.status == Conf.WORKING:
+                    status = term.green(str(Conf.WORKING))
+                elif stat.status == Conf.STOPPING:
+                    status = term.yellow(str(Conf.STOPPING))
+                elif stat.status == Conf.STOPPED:
+                    status = term.red(str(Conf.STOPPED))
+                elif stat.status == Conf.IDLE:
+                    status = str(Conf.IDLE)
+                # color q's
+                tasks = str(stat.task_q_size)
+                if stat.task_q_size > 0:
+                    tasks = term.cyan(str(stat.task_q_size))
+                    if Conf.QUEUE_LIMIT and stat.task_q_size == Conf.QUEUE_LIMIT:
+                        tasks = term.green(str(stat.task_q_size))
+                results = stat.done_q_size
+                if results > 0:
+                    results = term.cyan(str(results))
+                # color workers
+                workers = len(stat.workers)
+                if workers < Conf.WORKERS:
+                    workers = term.yellow(str(workers))
+                # format uptime
+                uptime = (timezone.now() - stat.tob).total_seconds()
+                hours, remainder = divmod(uptime, 3600)
+                minutes, seconds = divmod(remainder, 60)
+                uptime = "%d:%02d:%02d" % (hours, minutes, seconds)
+                # print to the terminal
+                print(
+                    term.move(i, 0)
+                    + term.center(stat.host[: col_width - 1], width=col_width - 1)
+                )
+                print(
+                    term.move(i, 1 * col_width)
+                    + term.center(str(stat.cluster_id)[-8:], width=col_width - 1)
+                )
+                print(
+                    term.move(i, 2 * col_width)
+                    + term.center(status, width=col_width - 1)
+                )
+                print(
+                    term.move(i, 3 * col_width)
+                    + term.center(workers, width=col_width - 1)
+                )
+                print(
+                    term.move(i, 4 * col_width)
+                    + term.center(tasks, width=col_width - 1)
+                )
+                print(
+                    term.move(i, 5 * col_width)
+                    + term.center(results, width=col_width - 1)
+                )
+                print(
+                    term.move(i, 6 * col_width)
+                    + term.center(stat.reincarnations, width=col_width - 1)
+                )
+                print(
+                    term.move(i, 7 * col_width)
+                    + term.center(uptime, width=col_width - 1)
+                )
+                i += 1
+            # bottom bar
+            i += 1
+            queue_size = broker.queue_size()
+            lock_size = broker.lock_size()
+            if lock_size:
+                queue_size = f"{queue_size}({lock_size})"
+            print(
+                term.move(i, 0)
+                + term.white_on_cyan(term.center(broker.info(), width=col_width * 2))
+            )
+            print(
+                term.move(i, 2 * col_width)
+                + term.black_on_cyan(term.center(_("Queued"), width=col_width))
+            )
+            print(
+                term.move(i, 3 * col_width)
+                + term.white_on_cyan(term.center(queue_size, width=col_width))
+            )
+            print(
+                term.move(i, 4 * col_width)
+                + term.black_on_cyan(term.center(_("Success"), width=col_width))
+            )
+            print(
+                term.move(i, 5 * col_width)
+                + term.white_on_cyan(
+                    term.center(models.Success.objects.count(), width=col_width)
+                )
+            )
+            print(
+                term.move(i, 6 * col_width)
+                + term.black_on_cyan(term.center(_("Failures"), width=col_width))
+            )
+            print(
+                term.move(i, 7 * col_width)
+                + term.white_on_cyan(
+                    term.center(models.Failure.objects.count(), width=col_width)
+                )
+            )
+            # for testing
+            if run_once:
+                return Stat.get_all(broker=broker)
+            print(term.move(i + 2, 0) + term.center(_("[Press q to quit]")))
+            val = term.inkey(timeout=1)
+
+
+def info(broker=None):
+    if not broker:
+        broker = get_broker()
+    try:
+        from blessed import Terminal
+
+        term = Terminal()
+    except ImportError:
+        print(BLESSED_INSTALL_MESSAGE)
+        return
+
+    broker.ping()
+    stat = Stat.get_all(broker=broker)
+    # general stats
+    clusters = len(stat)
+    workers = 0
+    reincarnations = 0
+    for cluster in stat:
+        workers += len(cluster.workers)
+        reincarnations += cluster.reincarnations
+    # calculate tasks pm and avg exec time
+    tasks_per = 0
+    per = _("day")
+    exec_time = 0
+    last_tasks = models.Success.objects.filter(
+        stopped__gte=timezone.now() - timedelta(hours=24)
+    )
+    tasks_per_day = last_tasks.count()
+    if tasks_per_day > 0:
+        # average execution time over the last 24 hours
+        if connection.vendor != "sqlite":
+            exec_time = last_tasks.aggregate(
+                time_taken=Sum(F("stopped") - F("started"))
+            )
+            exec_time = exec_time["time_taken"].total_seconds() / tasks_per_day
+        else:
+            # can't sum timedeltas on sqlite
+            for t in last_tasks:
+                exec_time += t.time_taken()
+            exec_time = exec_time / tasks_per_day
+        # tasks per second/minute/hour/day in the last 24 hours
+        if tasks_per_day > 24 * 60 * 60:
+            tasks_per = tasks_per_day / (24 * 60 * 60)
+            per = _("second")
+        elif tasks_per_day > 24 * 60:
+            tasks_per = tasks_per_day / (24 * 60)
+            per = _("minute")
+        elif tasks_per_day > 24:
+            tasks_per = tasks_per_day / 24
+            per = _("hour")
+        else:
+            tasks_per = tasks_per_day
+    # print to terminal
+    print(term.clear_eos())
+    col_width = int(term.width / 6)
+    print(
+        term.black_on_green(
+            term.center(
+                _("-- %(prefix)s %(version)s on %(info)s --")
+                % {
+                    "prefix": Conf.PREFIX.capitalize(),
+                    "version": ".".join(str(v) for v in VERSION),
+                    "info": broker.info(),
+                }
+            )
+        )
+    )
+    print(
+        term.cyan(_("Clusters"))
+        + term.move_x(1 * col_width)
+        + term.white(str(clusters))
+        + term.move_x(2 * col_width)
+        + term.cyan(_("Workers"))
+        + term.move_x(3 * col_width)
+        + term.white(str(workers))
+        + term.move_x(4 * col_width)
+        + term.cyan(_("Restarts"))
+        + term.move_x(5 * col_width)
+        + term.white(str(reincarnations))
+    )
+    print(
+        term.cyan(_("Queued"))
+        + term.move_x(1 * col_width)
+        + term.white(str(broker.queue_size()))
+        + term.move_x(2 * col_width)
+        + term.cyan(_("Successes"))
+        + term.move_x(3 * col_width)
+        + term.white(str(models.Success.objects.count()))
+        + term.move_x(4 * col_width)
+        + term.cyan(_("Failures"))
+        + term.move_x(5 * col_width)
+        + term.white(str(models.Failure.objects.count()))
+    )
+    print(
+        term.cyan(_("Schedules"))
+        + term.move_x(1 * col_width)
+        + term.white(str(models.Schedule.objects.count()))
+        + term.move_x(2 * col_width)
+        + term.cyan(_("Tasks/%(per)s") % {"per": per})
+        + term.move_x(3 * col_width)
+        + term.white(f"{tasks_per:.2f}")
+        + term.move_x(4 * col_width)
+        + term.cyan(_("Avg time"))
+        + term.move_x(5 * col_width)
+        + term.white(f"{exec_time:.4f}")
+    )
+    return True
+
+
+def memory(run_once=False, workers=False, broker=None):
+    if not broker:
+        broker = get_broker()
+    try:
+        from blessed import Terminal
+
+        term = Terminal()
+    except ImportError:
+        print(BLESSED_INSTALL_MESSAGE)
+        return
+    broker.ping()
+    if not psutil:
+        print(term.clear_eos())
+        print(
+            term.white_on_red(
+                'Cannot start "qmemory" command. Missing "psutil" library.'
+            )
+        )
+        return
+    with term.fullscreen(), term.hidden_cursor(), term.cbreak():
+        MEMORY_AVAILABLE_LOWEST_PERCENTAGE = 100.0
+        MEMORY_AVAILABLE_LOWEST_PERCENTAGE_AT = timezone.now()
+        cols = 8
+        val = None
+        start_width = int(term.width / cols)
+        while val not in ["q", "Q"]:
+            col_width = int(term.width / cols)
+            # In case of resize
+            if col_width != start_width:
+                print(term.clear())
+                start_width = col_width
+            # sentinel, monitor and workers memory usage
+            print(
+                term.move(0, 0 * col_width)
+                + term.black_on_green(term.center(_("Host"), width=col_width - 1))
+            )
+            print(
+                term.move(0, 1 * col_width)
+                + term.black_on_green(term.center(_("Id"), width=col_width - 1))
+            )
+            print(
+                term.move(0, 2 * col_width)
+                + term.black_on_green(
+                    term.center(_("Available (%)"), width=col_width - 1)
+                )
+            )
+            print(
+                term.move(0, 3 * col_width)
+                + term.black_on_green(
+                    term.center(_("Available (MB)"), width=col_width - 1)
+                )
+            )
+            print(
+                term.move(0, 4 * col_width)
+                + term.black_on_green(term.center(_("Total (MB)"), width=col_width - 1))
+            )
+            print(
+                term.move(0, 5 * col_width)
+                + term.black_on_green(
+                    term.center(_("Sentinel (MB)"), width=col_width - 1)
+                )
+            )
+            print(
+                term.move(0, 6 * col_width)
+                + term.black_on_green(
+                    term.center(_("Monitor (MB)"), width=col_width - 1)
+                )
+            )
+            print(
+                term.move(0, 7 * col_width)
+                + term.black_on_green(
+                    term.center(_("Workers (MB)"), width=col_width - 1)
+                )
+            )
+            row = 2
+            stats = Stat.get_all(broker=broker)
+            print(term.clear_eos())
+            for stat in stats:
+                # memory available (%)
+                memory_available_percentage = round(
+                    psutil.virtual_memory().available
+                    * 100
+                    / psutil.virtual_memory().total,
+                    2,
+                )
+                # memory available (MB)
+                memory_available = round(
+                    psutil.virtual_memory().available / 1024**2, 2
+                )
+                if memory_available_percentage < MEMORY_AVAILABLE_LOWEST_PERCENTAGE:
+                    MEMORY_AVAILABLE_LOWEST_PERCENTAGE = memory_available_percentage
+                    MEMORY_AVAILABLE_LOWEST_PERCENTAGE_AT = timezone.now()
+                print(
+                    term.move(row, 0 * col_width)
+                    + term.center(stat.host[: col_width - 1], width=col_width - 1)
+                )
+                print(
+                    term.move(row, 1 * col_width)
+                    + term.center(str(stat.cluster_id)[-8:], width=col_width - 1)
+                )
+                print(
+                    term.move(row, 2 * col_width)
+                    + term.center(memory_available_percentage, width=col_width - 1)
+                )
+                print(
+                    term.move(row, 3 * col_width)
+                    + term.center(memory_available, width=col_width - 1)
+                )
+                print(
+                    term.move(row, 4 * col_width)
+                    + term.center(
+                        round(psutil.virtual_memory().total / 1024**2, 2),
+                        width=col_width - 1,
+                    )
+                )
+                print(
+                    term.move(row, 5 * col_width)
+                    + term.center(get_process_mb(stat.sentinel), width=col_width - 1)
+                )
+                print(
+                    term.move(row, 6 * col_width)
+                    + term.center(
+                        get_process_mb(getattr(stat, "monitor", None)),
+                        width=col_width - 1,
+                    )
+                )
+                workers_mb = 0
+                for worker_pid in stat.workers:
+                    result = get_process_mb(worker_pid)
+                    if isinstance(result, str):
+                        result = 0
+                    workers_mb += result
+                print(
+                    term.move(row, 7 * col_width)
+                    + term.center(
+                        workers_mb or "NO_PROCESSES_FOUND", width=col_width - 1
+                    )
+                )
+                row += 1
+            # each worker's memory usage
+            if workers:
+                row += 2
+                col_width = int(term.width / (1 + Conf.WORKERS))
+                print(
+                    term.move(row, 0 * col_width)
+                    + term.black_on_cyan(term.center(_("Id"), width=col_width - 1))
+                )
+                for worker_num in range(Conf.WORKERS):
+                    print(
+                        term.move(row, (worker_num + 1) * col_width)
+                        + term.black_on_cyan(
+                            term.center(
+                                "Worker #{} (MB)".format(worker_num + 1),
+                                width=col_width - 1,
+                            )
+                        )
+                    )
+                row += 2
+                for stat in stats:
+                    print(
+                        term.move(row, 0 * col_width)
+                        + term.center(str(stat.cluster_id)[-8:], width=col_width - 1)
+                    )
+                    for idx, worker_pid in enumerate(stat.workers):
+                        mb_used = get_process_mb(worker_pid)
+                        print(
+                            term.move(row, (idx + 1) * col_width)
+                            + term.center(mb_used, width=col_width - 1)
+                        )
+                    row += 1
+            row += 1
+            print(
+                term.move(row, 0)
+                + _("Available lowest (): %(memory_percent)s ((at)s)")
+                % {
+                    "memory_percent": str(MEMORY_AVAILABLE_LOWEST_PERCENTAGE),
+                    "at": MEMORY_AVAILABLE_LOWEST_PERCENTAGE_AT.strftime(
+                        "%Y-%m-%d %H:%M:%S+00:00"
+                    ),
+                }
+            )
+            # for testing
+            if run_once:
+                return Stat.get_all(broker=broker)
+            print(term.move(row + 2, 0) + term.center(_("[Press q to quit]")))
+            val = term.inkey(timeout=1)
+
+
+def get_ids():
+    # prints id (PID) of running clusters
+    stat = Stat.get_all()
+    if stat:
+        for s in stat:
+            print(s.cluster_id)
+    else:
+        print(_("No clusters appear to be running."))
+    return True

django_q/pusher.py

@@ -0,0 +1,71 @@
+from multiprocessing import Event
+from multiprocessing.process import current_process
+from multiprocessing.queues import Queue
+from time import sleep
+
+from django import core
+from django.utils.translation import gettext_lazy as _
+from django.apps.registry import apps
+
+try:
+    apps.check_apps_ready()
+except core.exceptions.AppRegistryNotReady:
+    import django
+
+    django.setup()
+
+from django_q.brokers import Broker, get_broker
+from django_q.conf import Conf, logger
+from django_q.signing import BadSignature, SignedPackage
+
+try:
+    import setproctitle
+except ModuleNotFoundError:
+    setproctitle = None
+
+
+def pusher(task_queue: Queue, event: Event, broker: Broker = None):
+    """
+    Pulls tasks of the broker and puts them in the task queue
+    :type broker:
+    :type task_queue: multiprocessing.Queue
+    :type event: multiprocessing.Event
+    """
+    if not broker:
+        broker = get_broker()
+    proc_name = current_process().name
+    if setproctitle:
+        setproctitle.setproctitle(f"qcluster {proc_name} pusher")
+    logger.info(
+        _("%(name)s pushing tasks at %(id)s")
+        % {"name": proc_name, "id": current_process().pid}
+    )
+    while True:
+        try:
+            task_set = broker.dequeue()
+        except Exception:
+            logger.exception("Failed to pull task from broker")
+            # broker probably crashed. Let the sentinel handle it.
+            sleep(10)
+            break
+        if task_set:
+            for task in task_set:
+                ack_id = task[0]
+                # unpack the task
+                try:
+                    task = SignedPackage.loads(task[1])
+                except (TypeError, BadSignature):
+                    logger.exception("Failed to push task to queue")
+                    broker.fail(ack_id)
+                    continue
+                task[
+                    "cluster"
+                ] = Conf.CLUSTER_NAME  # save actual cluster name to orm task table
+                task["ack_id"] = ack_id
+                task_queue.put(task)
+            logger.debug(
+                _("queueing from %(list_key)s") % {"list_key": broker.list_key}
+            )
+        if event.is_set():
+            break
+    logger.info(_("%(name)s stopped pushing tasks") % {"name": current_process().name})

django_q/scheduler.py

@@ -0,0 +1,133 @@
+import ast
+from multiprocessing.process import current_process
+
+from django import core, db
+from django.utils import timezone
+from django.utils.translation import gettext_lazy as _
+
+from django.apps.registry import apps
+
+try:
+    apps.check_apps_ready()
+except core.exceptions.AppRegistryNotReady:
+    import django
+
+    django.setup()
+
+from django_q.brokers import Broker, get_broker
+from django_q.conf import Conf, logger
+from django_q.humanhash import humanize
+from django_q.models import Schedule
+from django_q.tasks import async_task
+from django_q.utils import close_old_django_connections, localtime
+
+
+def scheduler(broker: Broker = None):
+    """
+    Creates a task from a schedule at the scheduled time and schedules next run
+    """
+    if not broker:
+        broker = get_broker()
+    close_old_django_connections()
+    try:
+        # Only default cluster will handler schedule with default(null) cluster
+        Q_default = (
+            db.models.Q(cluster__isnull=True)
+            if Conf.CLUSTER_NAME == Conf.PREFIX
+            else db.models.Q(pk__in=[])
+        )
+
+        with db.transaction.atomic(using=db.router.db_for_write(Schedule)):
+            for s in (
+                Schedule.objects.select_for_update()
+                .exclude(repeats=0)
+                .filter(next_run__lt=timezone.now())
+                .filter(Q_default | db.models.Q(cluster=Conf.CLUSTER_NAME))
+            ):
+                args = ()
+                kwargs = {}
+                # get args, kwargs and hook
+                if s.kwargs:
+                    try:
+                        # first try the dict syntax
+                        kwargs = ast.literal_eval(s.kwargs)
+                    except (SyntaxError, ValueError):
+                        # else use the kwargs syntax
+                        try:
+                            parsed_kwargs = (
+                                ast.parse(f"f({s.kwargs})").body[0].value.keywords
+                            )
+                            kwargs = {
+                                kwarg.arg: ast.literal_eval(kwarg.value)
+                                for kwarg in parsed_kwargs
+                            }
+                        except (SyntaxError, ValueError):
+                            kwargs = {}
+                if s.args:
+                    args = ast.literal_eval(s.args)
+                    # single value won't eval to tuple, so:
+                    if type(args) != tuple:
+                        args = (args,)
+                q_options = kwargs.get("q_options", {})
+                if s.intended_date_kwarg:
+                    kwargs[s.intended_date_kwarg] = s.next_run.isoformat()
+                if s.hook:
+                    q_options["hook"] = s.hook
+                # set up the next run time
+                if s.schedule_type != s.ONCE:
+                    next_run = s.next_run
+                    while True:
+                        next_run = s.calculate_next_run(next_run)
+                        if Conf.CATCH_UP or next_run > localtime():
+                            break
+
+                    s.next_run = next_run
+                    s.repeats += -1
+                # send it to the cluster; any cluster name is allowed in multi-queue scenarios
+                # because `broker_name` is confusing, using `cluster` name is recommended and takes precedence
+                q_options["cluster"] = s.cluster or q_options.get(
+                    "cluster", q_options.pop("broker_name", None)
+                )
+                if (
+                    q_options["cluster"] is None
+                    or q_options["cluster"] == Conf.CLUSTER_NAME
+                ):
+                    q_options["broker"] = broker
+                q_options["group"] = q_options.get("group", s.name or s.id)
+                kwargs["q_options"] = q_options
+                s.task = async_task(s.func, *args, **kwargs)
+                # log it
+                if not s.task:
+                    logger.error(
+                        _(
+                            "%(process_name)s failed to create a task from schedule "
+                            "[%(schedule)s]"
+                        )
+                        % {
+                            "process_name": current_process().name,
+                            "schedule": s.name or s.id,
+                        }
+                    )
+                else:
+                    logger.info(
+                        _(
+                            "%(process_name)s created task %(task_name)s from schedule "
+                            "[%(schedule)s]"
+                        )
+                        % {
+                            "process_name": current_process().name,
+                            "task_name": humanize(s.task),
+                            "schedule": s.name or s.id,
+                        }
+                    )
+                # default behavior is to delete a ONCE schedule
+                if s.schedule_type == s.ONCE:
+                    if s.repeats < 0:
+                        s.delete()
+                        continue
+                    # but not if it has a positive repeats
+                    s.repeats = 0
+                # save the schedule
+                s.save()
+    except Exception:
+        logger.exception("Could not create task from schedule")

django_q/signals.py

@@ -31,6 +31,7 @@ def call_hook(sender, instance, **kwargs):
                 % {"hook": instance.hook, "name": instance.name, "error": str(e)}
             )
 
+
 # args: proc_name
 post_spawn = Signal()
 

django_q/tasks.py

@@ -148,7 +148,7 @@ def result(task_id, wait=0, cached=Conf.CACHED):
     start = time()
     while True:
         r = Task.get_result(task_id)
-        if r:
+        if r is not None:
             return r
         if (time() - start) * 1000 >= wait >= 0:
             break
@@ -763,7 +763,8 @@ class AsyncTask:
 
 def _sync(pack):
     """Simulate a package travelling through the cluster."""
-    from django_q.cluster import monitor, worker
+    from django_q.monitor import monitor
+    from django_q.worker import worker
 
     task_queue = Queue()
     result_queue = Queue()

django_q/tests/tasks.py

@@ -46,6 +46,10 @@ def hello():
     return "hello"
 
 
+def return_falsy_value():
+    return []
+
+
 def result(obj):
     print(f"RESULT HOOK {obj.name} : {obj.result()}")
 

django_q/tests/test_admin.py

@@ -64,7 +64,7 @@ def test_admin_views(admin_client, monkeypatch):
 
     # resubmit the failure
     url = reverse("admin:django_q_failure_changelist")
-    data = {"action": "retry_failed", "_selected_action": [f.pk]}
+    data = {"action": "resubmit_task", "_selected_action": [f.pk]}
     response = admin_client.post(url, data)
     assert response.status_code == 302
     assert Failure.objects.filter(name=f.id).exists() is False
@@ -84,3 +84,10 @@ def test_admin_views(admin_client, monkeypatch):
     data = {"post": "yes"}
     response = admin_client.post(url, data)
     assert response.status_code == 302
+    # Resubmit a successful task.
+    url = reverse("admin:django_q_success_changelist")
+    data = {"action": "resubmit_task", "_selected_action": [t.pk]}
+    initial_queue_count = OrmQ.objects.count()
+    response = admin_client.post(url, data)
+    assert response.status_code == 302
+    assert OrmQ.objects.count() > initial_queue_count

django_q/tests/test_cached.py

@@ -3,8 +3,9 @@ from multiprocessing import Event, Value
 import pytest
 
 from django_q.brokers import get_broker
-from django_q.cluster import monitor, pusher, worker
 from django_q.conf import Conf
+from django_q.monitor import monitor
+from django_q.pusher import pusher
 from django_q.queues import Queue
 from django_q.tasks import (
     AsyncTask,
@@ -21,6 +22,7 @@ from django_q.tasks import (
     result,
     result_group,
 )
+from django_q.worker import worker
 
 
 @pytest.fixture

django_q/tests/test_cluster.py

@@ -12,10 +12,12 @@ import pytest
 from django.utils import timezone
 
 from django_q.brokers import Broker, get_broker
-from django_q.cluster import Cluster, Sentinel, monitor, pusher, save_task, worker
+from django_q.cluster import Cluster, Sentinel
 from django_q.conf import Conf
 from django_q.humanhash import DEFAULT_WORDLIST, uuid
 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.status import Stat
@@ -29,8 +31,9 @@ from django_q.tasks import (
     result,
     result_group,
 )
-from django_q.tests.tasks import multiply, TaskError
+from django_q.tests.tasks import TaskError, multiply
 from django_q.utils import add_months, add_years
+from django_q.worker import worker
 
 myPath = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, myPath + "/../")
@@ -138,6 +141,30 @@ def test_cluster(broker):
     broker.delete_queue()
 
 
+@pytest.mark.django_db
+def test_results(broker):
+    broker.list_key = "cluster_test:q"
+    broker.delete_queue()
+    a = async_task(
+        "django_q.tests.tasks.return_falsy_value",
+        broker=broker,
+    )
+
+    task_queue = Queue()
+    stop_event = Event()
+    stop_event.set()
+    pusher(task_queue, stop_event, broker=broker)
+    task_queue.put("STOP")
+    result_queue = Queue()
+    worker(task_queue, result_queue, Value("f", -1))
+    result_queue.put("STOP")
+    monitor(result_queue)
+
+    # should not loop indefinitely when a real value is returned
+    value = result(a, wait=-1)
+    assert value == []
+
+
 @pytest.mark.django_db
 def test_enqueue(broker, admin_user):
     broker.list_key = "cluster_test:q"

django_q/tests/test_monitor.py

@@ -5,7 +5,7 @@ import pytest
 from django_q.brokers import get_broker
 from django_q.cluster import Cluster
 from django_q.conf import Conf
-from django_q.monitor import get_ids, info, monitor
+from django_q.monitor_terminal import get_ids, info, monitor
 from django_q.status import Stat
 from django_q.tasks import async_task
 

django_q/tests/test_scheduler.py

@@ -3,8 +3,8 @@ from datetime import datetime, timedelta
 from multiprocessing import Event, Value
 from unittest import mock
 
-import pytest
 import django
+import pytest
 from django.core.exceptions import ValidationError
 from django.db import IntegrityError
 from django.test import override_settings
@@ -12,9 +12,11 @@ from django.utils import timezone
 from django.utils.timezone import is_naive
 
 from django_q.brokers import Broker, get_broker
-from django_q.cluster import localtime, monitor, pusher, scheduler, worker
 from django_q.conf import Conf
+from django_q.monitor import monitor
+from django_q.pusher import pusher
 from django_q.queues import Queue
+from django_q.scheduler import scheduler
 from django_q.tasks import Schedule, fetch
 from django_q.tasks import schedule as create_schedule
 from django_q.tests.settings import BASE_DIR
@@ -22,7 +24,8 @@ from django_q.tests.testing_utilities.multiple_database_routers import (
     TestingMultipleAppsDatabaseRouter,
     TestingReplicaDatabaseRouter,
 )
-from django_q.utils import add_months
+from django_q.utils import add_months, localtime
+from django_q.worker import worker
 
 if django.VERSION < (4, 0):
     # pytz is the default in django 3.2. Remove when no support for 3.2
@@ -85,7 +88,7 @@ def test_scheduler_daylight_saving_time_daily(broker, monkeypatch):
     # 28th of March 2021 is the day when sunlight saving starts (at 2 am)
 
     monkeypatch.setattr(Conf, "TIME_ZONE", "Europe/Amsterdam")
-    tz = ZoneInfo('Europe/Amsterdam')
+    tz = ZoneInfo("Europe/Amsterdam")
     broker.list_key = "scheduler_test:q"
     # Let's start a schedule at 1 am on the 27th of March. This is in AMS timezone.
     # So, 2021-03-27 00:00:00 when saved (due to TZ being Amsterdam and saved in UTC)
@@ -181,7 +184,6 @@ def test_scheduler_daylight_saving_time_daily(broker, monkeypatch):
     assert str(next_run) == "2021-11-01 01:00:00+01:00"
 
 
-
 @pytest.mark.django_db
 def test_scheduler(broker, monkeypatch):
     broker.list_key = "scheduler_test:q"
@@ -407,7 +409,7 @@ def test_scheduler(broker, monkeypatch):
 def test_intended_schedule_kwarg(broker, monkeypatch):
     broker.list_key = "scheduler_test:q"
     broker.delete_queue()
-    run_date = timezone.now()-timedelta(hours=1)
+    run_date = timezone.now() - timedelta(hours=1)
     schedule = create_schedule(
         "math.copysign",
         1,
@@ -417,10 +419,10 @@ def test_intended_schedule_kwarg(broker, monkeypatch):
         schedule_type=Schedule.HOURLY,
         repeats=1,
         next_run=run_date,
-        intended_date_kwarg='intended_date',
+        intended_date_kwarg="intended_date",
     )
     assert schedule.last_run() is None
-    assert schedule.intended_date_kwarg == 'intended_date'
+    assert schedule.intended_date_kwarg == "intended_date"
     # run scheduler
     scheduler(broker=broker)
     # set up the workflow
@@ -431,8 +433,8 @@ def test_intended_schedule_kwarg(broker, monkeypatch):
     pusher(task_queue, stop_event, broker=broker)
     assert task_queue.qsize() == 1
     task = task_queue.get()
-    assert 'intended_date' in task['kwargs']
+    assert "intended_date" in task["kwargs"]
-    assert task['kwargs']['intended_date'] == run_date.isoformat()
+    assert task["kwargs"]["intended_date"] == run_date.isoformat()
 
 
 @override_settings(

django_q/utils.py

@@ -1,13 +1,13 @@
-from datetime import datetime
 import calendar
 import inspect
-from datetime import date
+from datetime import date, datetime
 
 import django
-from django.utils import timezone
+from django import db
 from django.conf import settings
+from django.utils import timezone
 
-from django_q.conf import Conf
+from django_q.conf import Conf, logger
 
 if django.VERSION < (4, 0):
     # pytz is the default in django 3.2. Remove when no support for 3.2
@@ -60,7 +60,7 @@ def get_func_repr(func):
 def localtime(value=None) -> datetime:
     """Override for timezone.localtime to deal with naive times and local times"""
     if settings.USE_TZ:
-        if django.VERSION >= (4, 0) and settings.USE_DEPRECATED_PYTZ:
+        if django.VERSION >= (4, 0) and getattr(settings, "USE_DEPRECATED_PYTZ", False):
             import pytz
 
             convert_to_tz = pytz.timezone(Conf.TIME_ZONE)
@@ -72,3 +72,17 @@ def localtime(value=None) -> datetime:
         return datetime.now()
     else:
         return value
+
+
+def close_old_django_connections():
+    """
+    Close django connections unless running with sync=True.
+    """
+    if Conf.SYNC:
+        logger.warning(
+            "Preserving django database connections because sync=True. Beware "
+            "that tasks are now injected in the calling context/transactions "
+            "which may result in unexpected behaviour."
+        )
+    else:
+        db.close_old_connections()

django_q/worker.py

@@ -0,0 +1,127 @@
+import pydoc
+import traceback
+from multiprocessing import Value
+from multiprocessing.process import current_process
+from multiprocessing.queues import Queue
+
+from django import core
+from django.utils import timezone
+from django.utils.translation import gettext_lazy as _
+from django.apps.registry import apps
+
+try:
+    apps.check_apps_ready()
+except core.exceptions.AppRegistryNotReady:
+    import django
+
+    django.setup()
+
+from django_q.conf import Conf, error_reporter, logger, resource, setproctitle
+from django_q.signals import post_spawn, pre_execute
+from django_q.utils import close_old_django_connections, get_func_repr
+
+try:
+    import psutil
+except ImportError:
+    psutil = None
+
+try:
+    import setproctitle
+except ModuleNotFoundError:
+    setproctitle = None
+
+
+def worker(
+    task_queue: Queue, result_queue: Queue, timer: Value, timeout: int = Conf.TIMEOUT
+):
+    """
+    Takes a task from the task queue, tries to execute it and puts the result back in
+    the result queue
+    :param timeout: number of seconds wait for a worker to finish.
+    :type task_queue: multiprocessing.Queue
+    :type result_queue: multiprocessing.Queue
+    :type timer: multiprocessing.Value
+    """
+    proc_name = current_process().name
+    logger.info(
+        _("%(proc_name)s ready for work at %(id)s")
+        % {"proc_name": proc_name, "id": current_process().pid}
+    )
+    post_spawn.send(sender="django_q", proc_name=proc_name)
+    if setproctitle:
+        setproctitle.setproctitle(f"qcluster {proc_name} idle")
+    task_count = 0
+    if timeout is None:
+        timeout = -1
+    # Start reading the task queue
+    for task in iter(task_queue.get, "STOP"):
+        result = None
+        timer.value = -1  # Idle
+        task_count += 1
+        f = task["func"]
+
+        # Log task creation and set process name
+        # Get the function from the task
+        func_name = get_func_repr(f)
+        task_name = task["name"]
+        task_desc = _("%(proc_name)s processing %(task_name)s '%(func_name)s'") % {
+            "proc_name": proc_name,
+            "func_name": func_name,
+            "task_name": task_name,
+        }
+        if "group" in task:
+            task_desc += f" [{task['group']}]"
+        logger.info(task_desc)
+
+        if setproctitle:
+            proc_title = f"qcluster {proc_name} processing {task_name} '{func_name}'"
+            if "group" in task:
+                proc_title += f" [{task['group']}]"
+            setproctitle.setproctitle(proc_title)
+
+        # if it's not an instance try to get it from the string
+        if not callable(f):
+            # locate() returns None if f cannot be loaded
+            f = pydoc.locate(f)
+        close_old_django_connections()
+        timer_value = task.pop("timeout", timeout)
+        # signal execution
+        pre_execute.send(sender="django_q", func=f, task=task)
+        # execute the payload
+        timer.value = timer_value  # Busy
+
+        try:
+            if f is None:
+                # raise a meaningfull error if task["func"] is not a valid function
+                raise ValueError(f"Function {task['func']} is not defined")
+            res = f(*task["args"], **task["kwargs"])
+            result = (res, True)
+        except Exception as e:
+            result = (f"{e} : {traceback.format_exc()}", False)
+            if error_reporter:
+                error_reporter.report()
+            if task.get("sync", False):
+                raise
+        with timer.get_lock():
+            # Process result
+            task["result"] = result[0]
+            task["success"] = result[1]
+            task["stopped"] = timezone.now()
+            result_queue.put(task)
+            timer.value = -1  # Idle
+            if setproctitle:
+                setproctitle.setproctitle(f"qcluster {proc_name} idle")
+            # Recycle
+            if task_count == Conf.RECYCLE or rss_check():
+                timer.value = -2  # Recycled
+                break
+    logger.info(_("%(proc_name)s stopped doing work") % {"proc_name": proc_name})
+
+
+def rss_check():
+    if Conf.MAX_RSS:
+        if resource:
+            return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss >= Conf.MAX_RSS
+        elif psutil:
+            return psutil.Process().memory_info().rss >= Conf.MAX_RSS * 1024
+    return False

docs/cluster.rst

@@ -11,36 +11,36 @@ Start your cluster using Django's ``manage.py`` command::
 
 You should see the cluster starting ::
 
-    10:57:40 [Q] INFO Q Cluster-31781 starting.
+    10:57:40 [Q] INFO Q Cluster freddie-uncle-twenty-ten starting.
-    10:57:40 [Q] INFO Process-1:1 ready for work at 31784
+    10:57:40 [Q] INFO Process-ede257774c4444c980ab479f10947acc ready for work at 31784
-    10:57:40 [Q] INFO Process-1:2 ready for work at 31785
+    10:57:40 [Q] INFO Process-ed580482da3f42968230baa2e4253e42 ready for work at 31785
-    10:57:40 [Q] INFO Process-1:3 ready for work at 31786
+    10:57:40 [Q] INFO Process-8a370dc2bc1d49aa9864e517c9895f74 ready for work at 31786
-    10:57:40 [Q] INFO Process-1:4 ready for work at 31787
+    10:57:40 [Q] INFO Process-74912f9844264d1397c6e54476b530c0 ready for work at 31787
-    10:57:40 [Q] INFO Process-1:5 ready for work at 31788
+    10:57:40 [Q] INFO Process-b00edb26c6074a6189e5696c60aeb35b ready for work at 31788
-    10:57:40 [Q] INFO Process-1:6 ready for work at 31789
+    10:57:40 [Q] INFO Process-b0862965db04479f9784a26639ee51e0 ready for work at 31789
-    10:57:40 [Q] INFO Process-1:7 ready for work at 31790
+    10:57:40 [Q] INFO Process-7e8abbb8ca2d4d9bb20a937dd5e2872b ready for work at 31790
-    10:57:40 [Q] INFO Process-1:8 ready for work at 31791
+    10:57:40 [Q] INFO Process-b0862965db04479f9784a26639ee51e0 ready for work at 31791
-    10:57:40 [Q] INFO Process-1:9 monitoring at 31792
+    10:57:40 [Q] INFO Process-67fa9461ac034736a766cd813f617e62 monitoring at 31792
-    10:57:40 [Q] INFO Process-1 guarding cluster at 31783
+    10:57:40 [Q] INFO Process-eac052c646b2459797cee98bdb84c85d guarding cluster at 31783
-    10:57:40 [Q] INFO Process-1:10 pushing tasks at 31793
+    10:57:40 [Q] INFO Process-5d98deb19b1e4b2da2ef1e5bd6824f75 pushing tasks at 31793
-    10:57:40 [Q] INFO Q Cluster-31781 running.
+    10:57:40 [Q] INFO Q Cluster freddie-uncle-twenty-ten running.
 
 
 Stopping the cluster with ctrl-c or either the ``SIGTERM`` and ``SIGKILL`` signals, will initiate the :ref:`stop_procedure`::
 
-    16:44:12 [Q] INFO Q Cluster-31781 stopping.
+    16:44:12 [Q] INFO Q Cluster freddie-uncle-twenty-ten stopping.
-    16:44:12 [Q] INFO Process-1 stopping cluster processes
+    16:44:12 [Q] INFO Process-eac052c646b2459797cee98bdb84c85d stopping cluster processes
-    16:44:13 [Q] INFO Process-1:10 stopped pushing tasks
+    16:44:13 [Q] INFO Process-5d98deb19b1e4b2da2ef1e5bd6824f75 stopped pushing tasks
-    16:44:13 [Q] INFO Process-1:6 stopped doing work
+    16:44:13 [Q] INFO Process-b0862965db04479f9784a26639ee51e0 stopped doing work
-    16:44:13 [Q] INFO Process-1:4 stopped doing work
+    16:44:13 [Q] INFO Process-7e8abbb8ca2d4d9bb20a937dd5e2872b stopped doing work
-    16:44:13 [Q] INFO Process-1:1 stopped doing work
+    16:44:13 [Q] INFO Process-b0862965db04479f9784a26639ee51e0 stopped doing work
-    16:44:13 [Q] INFO Process-1:5 stopped doing work
+    16:44:13 [Q] INFO Process-b00edb26c6074a6189e5696c60aeb35b stopped doing work
-    16:44:13 [Q] INFO Process-1:7 stopped doing work
+    16:44:13 [Q] INFO Process-74912f9844264d1397c6e54476b530c0 stopped doing work
-    16:44:13 [Q] INFO Process-1:3 stopped doing work
+    16:44:13 [Q] INFO Process-8a370dc2bc1d49aa9864e517c9895f74 stopped doing work
-    16:44:13 [Q] INFO Process-1:8 stopped doing work
+    16:44:13 [Q] INFO Process-ed580482da3f42968230baa2e4253e42 stopped doing work
-    16:44:13 [Q] INFO Process-1:2 stopped doing work
+    16:44:13 [Q] INFO Process-ede257774c4444c980ab479f10947acc stopped doing work
-    16:44:14 [Q] INFO Process-1:9 stopped monitoring results
+    16:44:14 [Q] INFO Process-67fa9461ac034736a766cd813f617e62 stopped monitoring results
-    16:44:15 [Q] INFO Q Cluster-31781 has stopped.
+    16:44:15 [Q] INFO Q Cluster freddie-uncle-twenty-ten has stopped.
 
 The number of workers, optional timeouts, recycles and cpu_affinity can be controlled via the :doc:`configure` settings.
 

docs/conf.py

@@ -73,9 +73,9 @@ author = "Ilan Steemers, Stan Triepels"
 # built documents.
 #
 # The short X.Y version.
-version = "1.5"
+version = "1.6"
 # The full version, including alpha/beta/rc tags.
-release = "1.5.2"
+release = "1.6.0"
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

docs/configure.rst

@@ -348,7 +348,7 @@ To use MongoDB as a message broker you simply provide the connection information
         }
     }
 
-The ``mongo`` dictionary can contain any of the parameters exposed by pymongo's `MongoClient <https://api.mongodb.org/python/current/api/pymongo/mongo_client.html#pymongo.mongo_client.MongoClient>`__
+The ``mongo`` dictionary can contain any of the parameters exposed by pymongo's `MongoClient <https://pymongo.readthedocs.io/en/stable/api/pymongo/mongo_client.html#pymongo.mongo_client.MongoClient>`__
 If you want to use a mongodb uri, you can supply it as the ``host`` parameter.
 
 mongo_db

docs/examples.rst

@@ -328,7 +328,7 @@ Requires cache to be enabled. Save file in your Django project's root directory
     # All django stuff has to come after the setup:
     django.setup()
 
-    from django_q.monitor import Stat
+    from django_q.status import Stat
     from django_q.conf import Conf
 
     # Set host and port settings

docs/index.rst

@@ -27,7 +27,7 @@ Features
 -  Rollbar and Sentry support
 
 
-Django Q2 is tested with: Python 3.8, 3.9, 3.10 and 3.11. Works with Django 3.2.x, 4.1.x and 4.2.x
+Django Q2 is tested with: Python 3.8, 3.9, 3.10, 3.11 and 3.12. Works with Django 3.2.x, 4.1.x, 4.2.x and 5.0.x
 
 Currently available in English, German and French.
 

docs/install.rst

@@ -24,15 +24,31 @@ Installation
 
     $ python manage.py qcluster
 
+
+Migrate from Django-Q to Django-Q2
+----------------------------------
+
+If you have an application with django-q running right now, you can simply swap the libraries and you should be good to go.::
+
+
+    $ pip uninstall django-q # you might have to uninstall django-q add-ons as well
+    $ pip install django-q2
+
+
+Then migrate the database to get the latest tables/fields::
+
+    $ python manage.py migrate
+
+
 Requirements
 ------------
 
-Django Q2 is tested for Python 3.8, 3.9, 3.10 and 3.11
+Django Q2 is tested for Python 3.8, 3.9, 3.10, 3.11 and 3.12
 
 -  `Django <https://www.djangoproject.com>`__
 
     Django Q2 aims to use as much of Django's standard offerings as possible.
-    The code is tested against Django versions `3.2.x`, `4.1.x` and`4.2.x`.
+    The code is tested against Django versions `3.2.x`, `4.1.x`, `4.2.x` and `5.0.x`.
 
 -  `Django-picklefield <https://github.com/gintas/django-picklefield>`__
 
@@ -99,11 +115,11 @@ Add-ons
 -------
 - `django-q-rollbar <https://github.com/danielwelch/django-q-rollbar>`__ is a Rollbar error reporter::
 
-    $ pip install django-q[rollbar]
+    $ pip install django-q2[rollbar]
 
 - `django-q-sentry <https://github.com/danielwelch/django-q-sentry>`__ is a Sentry error reporter::
 
-    $ pip install django-q[sentry]
+    $ pip install django-q2[sentry]
 
 - `django-q-email <https://github.com/joeyespo/django-q-email>`__ is a compatible Django email backend that will automatically async queue your emails.
 
@@ -129,7 +145,7 @@ Other known issues are:
 
 Python
 ~~~~~~
-Current tests are performed with 3.8, 3.9, 3.10 and 3.11
+Current tests are performed with 3.8, 3.9, 3.10, 3.11 and 3.12
 If you do encounter any regressions with earlier versions, please submit an issue on `github <https://github.com/GDay/django-q2>`__
 
 Open-source packages
@@ -139,8 +155,8 @@ You can reference the `requirements <https://github.com/GDay/django-q2/blob/mast
 
 Django
 ~~~~~~
-We strive to be compatible with last two major version of Django.
+We strive to be compatible with the last two major version of Django.
-At the moment this means we support the 3.2.x, 4.1.x and 4.2.x releases.
+At the moment this means we support the 3.2.x, 4.1.x, 4.2.x and 5.0.x releases.
 
-Since we are now no longer supporting Python 2, we can also not support older versions of Django that do not support Python >= 3.6
+Since we are now no longer supporting Python 2, we can also not support older versions of Django that do not support Python >= 3.8
 For this you can always use older releases, but they are no longer maintained.

docs/monitor.rst

@@ -111,7 +111,7 @@ You can check the status of your clusters straight from your code with the :clas
 
 .. code:: python
 
-    from django_q.monitor import Stat
+    from django_q.status import Stat
 
     for stat in Stat.get_all():
         print(stat.cluster_id, stat.status)

docs/signals.rst

@@ -13,6 +13,12 @@ Before enqueuing a task
 The ``django_q.signals.pre_enqueue`` signal is emitted before a task is
 enqueued. The task dictionary is given as the ``task`` argument.
 
+After spawning a worker process
+"""""""""""""""""""""""""""""""
+
+The ``django_q.signals.post_spawn`` signal is emitted after a worker process has
+spawned. The process name is given as the ``proc_name`` argument (string).
+
 Before executing a task
 """""""""""""""""""""""
 
@@ -37,7 +43,7 @@ Connecting to a Django Q2 signal is done the same as any other Django
 signal::
 
     from django.dispatch import receiver
-    from django_q.signals import pre_enqueue, pre_execute, post_execute
+    from django_q.signals import pre_enqueue, pre_execute, post_execute, post_spawn
 
     @receiver(pre_enqueue)
     def my_pre_enqueue_callback(sender, task, **kwargs):
@@ -51,4 +57,8 @@ signal::
     def my_post_execute_callback(sender, 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")
+
 

docs/tasks.rst

@@ -121,6 +121,7 @@ Optionally you can use the :class:`AsyncTask` class to instantiate a task and ke
     a.run()
 
     # wait indefinitely for the result and print it
+    # don't let the task return `None` or it will wait indefinitely
     print(a.result(wait=-1))
 
     # change the args
@@ -264,7 +265,7 @@ Reference
     Gets the result of a previously executed task
 
     :param str task_id: the uuid or name of the task
-    :param int wait: optional milliseconds to wait for a result. -1 for indefinite
+    :param int wait: optional milliseconds to wait for a result. -1 for indefinite, but be sure the result will not be `None` otherwise it will wait indefinitely!
     :param bool cached: run this against the cache backend.
     :returns: The result of the executed task
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "django-q2"
-version = "1.5.2"
+version = "1.6.0"
 packages = [
     { include = "django_q" },
 ]
@@ -30,6 +30,7 @@ classifiers = [
     'Programming Language :: Python :: 3.9',
     'Programming Language :: Python :: 3.10',
     'Programming Language :: Python :: 3.11',
+    'Programming Language :: Python :: 3.12',
     'Topic :: Internet :: WWW/HTTP',
     'Topic :: System :: Distributed Computing',
     'Topic :: Software Development :: Libraries :: Python Modules',
@@ -44,8 +45,11 @@ include = ['CHANGELOG.md']
 
 
 [tool.poetry.dependencies]
-python = ">=3.8, <4"
+python = ">=3.8,<4"
-django = ">=3.2"
+django = [
+    { version = ">=3.2, <5", python = ">= 3.8" },
+    { version = "^5.0a1", python = ">= 3.10", allow-prereleases = true }
+]
 django-picklefield = "^3.1"
 
 blessed = { version = "^1.19.1", optional = true }
@@ -55,11 +59,12 @@ django-redis = { version = "^5.2.0", optional = true }
 iron-mq = { version = "^0.9", optional = true }
 boto3 = { version = "^1.24.92", optional = true }
 pymongo = { version = "^4.2.0", optional = true }
-croniter = { version = "^1.3.7", optional = true }
+croniter = { version = "^2.0.1", optional = true }
 django-q-rollbar = {version = ">=0.1", optional = true}
 django-q-sentry = {version = ">=0.1", optional = true}
 redis = {version = "^4.3.4", optional = true}
 setproctitle = {version = "^1.3.2", optional = true}
+importlib-metadata = {version = ">=3.6", python = "<3.10"}
 
 
 [tool.poetry.dev-dependencies]