implement queued_jobs, fix #145 (#146)

Author: samuelcolvin

HISTORY.rst

@@ -3,10 +3,11 @@
 History
 -------
 
-v0.17.0 (unreleased)
-....................
+v0.17 (unreleased)
+..................
 * add ``worker.queue_read_limit``, fix #141, by @rubik
 * custom serializers, eg. to use msgpack rather than pickle, #143 by @rubik
+* add ``ArqRedis.queued_jobs`` utility method for getting queued jobs while testing, fix #145 by @samuelcolvin
 
 v0.16.1 (2019-08-02)
 ....................

arq/connections.py

@@ -11,7 +11,7 @@
 from aioredis import MultiExecError, Redis
 
 from .constants import default_queue_name, job_key_prefix, result_key_prefix
-from .jobs import Deserializer, Job, JobResult, Serializer, serialize_job
+from .jobs import Deserializer, Job, JobDef, JobResult, Serializer, deserialize_job, serialize_job
 from .utils import timestamp_ms, to_ms, to_unix_ms
 
 logger = logging.getLogger('arq.connections')
@@ -126,7 +126,7 @@ async def enqueue_job(
                 return
         return Job(job_id, redis=self, _deserializer=self.job_deserializer)
 
-    async def _get_job_result(self, key):
+    async def _get_job_result(self, key) -> JobResult:
         job_id = key[len(result_key_prefix) :]
         job = Job(job_id, self, _deserializer=self.job_deserializer)
         r = await job.result_info()
@@ -141,6 +141,19 @@ async def all_job_results(self) -> List[JobResult]:
         results = await asyncio.gather(*[self._get_job_result(k) for k in keys])
         return sorted(results, key=attrgetter('enqueue_time'))
 
+    async def _get_job_def(self, job_id, score) -> JobDef:
+        v = await self.get(job_key_prefix + job_id, encoding=None)
+        jd = deserialize_job(v, deserializer=self.job_deserializer)
+        jd.score = score
+        return jd
+
+    async def queued_jobs(self, *, queue_name: str = default_queue_name) -> List[JobDef]:
+        """
+        Get information about queued, mostly useful when testing.
+        """
+        jobs = await self.zrange(queue_name, withscores=True)
+        return await asyncio.gather(*[self._get_job_def(job_id, score) for job_id, score in jobs])
+
 
 async def create_pool(
     settings: RedisSettings = None,

tests/test_main.py

@@ -1,15 +1,16 @@
 import asyncio
+import dataclasses
 from collections import Counter
 from datetime import datetime
 from random import shuffle
 from time import time
 
 import pytest
-from pytest_toolbox.comparison import CloseToNow
+from pytest_toolbox.comparison import AnyInt, CloseToNow
 
 from arq.connections import ArqRedis
 from arq.constants import default_queue_name
-from arq.jobs import Job, SerializationError
+from arq.jobs import Job, JobDef, SerializationError
 from arq.utils import timestamp_ms
 from arq.worker import Retry, Worker, func
 
@@ -162,3 +163,42 @@ async def foobar(ctx):
     await w.main()
     with pytest.raises(SerializationError):
         await j1.result(pole_delay=0)
+
+
+async def test_get_jobs(arq_redis: ArqRedis):
+    await arq_redis.enqueue_job('foobar', a=1, b=2, c=3)
+    await asyncio.sleep(0.01)
+    await arq_redis.enqueue_job('second', 4, b=5, c=6)
+    await asyncio.sleep(0.01)
+    await arq_redis.enqueue_job('third', 7, b=8)
+    jobs = await arq_redis.queued_jobs()
+    assert [dataclasses.asdict(j) for j in jobs] == [
+        {
+            'function': 'foobar',
+            'args': (),
+            'kwargs': {'a': 1, 'b': 2, 'c': 3},
+            'job_try': None,
+            'enqueue_time': CloseToNow(),
+            'score': AnyInt(),
+        },
+        {
+            'function': 'second',
+            'args': (4,),
+            'kwargs': {'b': 5, 'c': 6},
+            'job_try': None,
+            'enqueue_time': CloseToNow(),
+            'score': AnyInt(),
+        },
+        {
+            'function': 'third',
+            'args': (7,),
+            'kwargs': {'b': 8},
+            'job_try': None,
+            'enqueue_time': CloseToNow(),
+            'score': AnyInt(),
+        },
+    ]
+    assert jobs[0].score < jobs[1].score < jobs[2].score
+    assert isinstance(jobs[0], JobDef)
+    assert isinstance(jobs[1], JobDef)
+    assert isinstance(jobs[2], JobDef)