adds queue_size()

returns the size of the current redis queue

django_q/__init__.py

@@ -4,7 +4,7 @@ import sys
 myPath = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, myPath)
 
-from .tasks import async, schedule, result, result_group, fetch, fetch_group, count_group, delete_group
+from .tasks import async, schedule, result, result_group, fetch, fetch_group, count_group, delete_group, queue_size
 from .models import Task, Schedule, Success, Failure
 from .cluster import Cluster
 from .monitor import Stat

django_q/tasks.py

@@ -152,6 +152,19 @@ def delete_group(group_id, tasks=False):
     return Task.delete_group(group_id, tasks)
 
 
+def queue_size(list_key=Conf.Q_LIST, r=redis_client):
+    """
+    Returns the current queue size.
+    Note that this doesn't count any tasks currently being processed by workers.
+
+    :param list_key: optional redis key
+    :param r: optional redis connection
+    :return: current queue size
+    :rtype: int
+    """
+    return r.llen(list_key)
+
+
 def _sync(pack):
     """Simulate a package travelling through the cluster."""
     task_queue = Queue()

django_q/tests/test_cluster.py

@@ -11,7 +11,7 @@ sys.path.insert(0, myPath + '/../')
 
 from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor
 from django_q.humanhash import DEFAULT_WORDLIST
-from django_q.tasks import fetch, fetch_group, async, result, result_group, count_group, delete_group
+from django_q.tasks import fetch, fetch_group, async, result, result_group, count_group, delete_group, queue_size
 from django_q.models import Task, Success
 from django_q.conf import Conf, redis_client
 from django_q.monitor import Stat
@@ -77,7 +77,7 @@ def test_cluster(r):
     list_key = 'cluster_test:q'
     r.delete(list_key)
     task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, list_key=list_key)
-    assert r.llen(list_key) == 1
+    assert queue_size(list_key=list_key, r=r) == 1
     task_queue = Queue()
     assert task_queue.qsize() == 0
     result_queue = Queue()
@@ -87,7 +87,7 @@ def test_cluster(r):
     # Test push
     pusher(task_queue, event, list_key=list_key, r=r)
     assert task_queue.qsize() == 1
-    assert r.llen(list_key) == 0
+    assert queue_size(list_key=list_key, r=r) == 0
     # Test work
     task_queue.put('STOP')
     worker(task_queue, result_queue, Value('f', -1))
@@ -142,14 +142,14 @@ def test_async(r, admin_user):
     assert isinstance(k, str)
     # run the cluster to execute the tasks
     task_count = 10
-    assert r.llen(list_key) == task_count
+    assert queue_size(list_key=list_key, r=r) == task_count
     task_queue = Queue()
     stop_event = Event()
     stop_event.set()
     # push the tasks
     for i in range(task_count):
         pusher(task_queue, stop_event, list_key=list_key, r=r)
-    assert r.llen(list_key) == 0
+    assert queue_size(list_key=list_key, r=r) == 0
     assert task_queue.qsize() == task_count
     task_queue.put('STOP')
     # let a worker handle them

django_q/tests/test_scheduler.py

@@ -6,7 +6,7 @@ from django.utils import timezone
 
 from django_q.conf import redis_client
 from django_q.cluster import pusher, worker, monitor, scheduler
-from django_q.tasks import Schedule, fetch, schedule as create_schedule
+from django_q.tasks import Schedule, fetch, schedule as create_schedule, queue_size
 
 
 @pytest.fixture
@@ -34,7 +34,7 @@ def test_scheduler(r):
     # push it
     pusher(task_queue, stop_event, list_key=list_key, r=r)
     assert task_queue.qsize() == 1
-    assert r.llen(list_key) == 0
+    assert queue_size(list_key=list_key,r=r) == 0
     task_queue.put('STOP')
     # let a worker handle them
     result_queue = Queue()

docs/tasks.rst

@@ -213,6 +213,14 @@ Reference
 
     Renamed from get_task
 
+.. py:function:: queue_size()
+
+    Returns the size of the broker queue.
+    Note that this does not count tasks currently being processed.
+
+    :returns: The amount of task packages in the broker
+    :rtype: int
+
 .. py:function:: result_group(group_id, failures=False)
 
     Returns the results of a task group