Raise ResultNotFound when Job.result() finds no job and no result

arq/jobs.py

@@ -18,6 +18,10 @@
 Deserializer = Callable[[bytes], Dict[str, Any]]
 
 
+class ResultNotFound(RuntimeError):
+    pass
+
+
 class JobStatus(str, Enum):
     """
     Enum of job statuses.
@@ -82,8 +86,10 @@ async def result(
         self, timeout: Optional[float] = None, *, poll_delay: float = 0.5, pole_delay: float = None
     ) -> Any:
         """
-        Get the result of the job, including waiting if it's not yet available. If the job raised an exception,
-        it will be raised here.
+        Get the result of the job or, if the job raised an exception, reraise it.
+
+        This function waits for the result if it's not yet available and the job is
+        present in the queue. Otherwise ResultNotFound is raised.
 
         :param timeout: maximum time to wait for the job result before raising ``TimeoutError``, will wait forever
         :param poll_delay: how often to poll redis for the job result
@@ -96,15 +102,24 @@ async def result(
             poll_delay = pole_delay
 
         async for delay in poll(poll_delay):
-            info = await self.result_info()
-            if info:
-                result = info.result
+            async with self._redis.pipeline(transaction=True):
+                v = await self._redis.get(result_key_prefix + self.job_id)
+                s = await self._redis.zscore(self._queue_name, self.job_id)
+
+            if v:
+                info = deserialize_result(v, deserializer=self._deserializer)
                 if info.success:
-                    return result
-                elif isinstance(result, (Exception, asyncio.CancelledError)):
-                    raise result
+                    return info.result
+                elif isinstance(info.result, (Exception, asyncio.CancelledError)):
+                    raise info.result
                 else:
-                    raise SerializationError(result)
+                    raise SerializationError(info.result)
+            elif s is None:
+                raise ResultNotFound(
+                    'Not waiting for job result because the job is not in queue. '
+                    'Is the worker function configured to keep result?'
+                )
+
             if timeout is not None and delay > timeout:
                 raise asyncio.TimeoutError()
 
@@ -169,6 +184,9 @@ async def abort(self, *, timeout: Optional[float] = None, poll_delay: float = 0.
             await self.result(timeout=timeout, poll_delay=poll_delay)
         except asyncio.CancelledError:
             return True
+        except ResultNotFound:
+            # We do not know if the job was cancelled or not
+            return False
         else:
             return False
 

tests/test_jobs.py

@@ -7,7 +7,15 @@
 from arq import Worker, func
 from arq.connections import ArqRedis, RedisSettings, create_pool
 from arq.constants import default_queue_name, in_progress_key_prefix, job_key_prefix, result_key_prefix
-from arq.jobs import DeserializationError, Job, JobResult, JobStatus, deserialize_job_raw, serialize_result
+from arq.jobs import (
+    DeserializationError,
+    Job,
+    JobResult,
+    JobStatus,
+    ResultNotFound,
+    deserialize_job_raw,
+    serialize_result,
+)
 
 
 async def test_job_in_progress(arq_redis: ArqRedis):
@@ -25,11 +33,36 @@ async def test_unknown(arq_redis: ArqRedis):
 
 
 async def test_result_timeout(arq_redis: ArqRedis):
-    j = Job('foobar', arq_redis)
+    j = await arq_redis.enqueue_job('foobar')
     with pytest.raises(asyncio.TimeoutError):
         await j.result(0.1, poll_delay=0)
 
 
+async def test_result_not_found(arq_redis: ArqRedis):
+    j = Job('foobar', arq_redis)
+    with pytest.raises(ResultNotFound):
+        await j.result()
+
+
+async def test_result_when_job_does_not_keep_result(arq_redis: ArqRedis, worker):
+    async def foobar(ctx):
+        pass
+
+    worker: Worker = worker(functions=[func(foobar, name='foobar', keep_result=0)])
+    j = await arq_redis.enqueue_job('foobar')
+
+    result_call = asyncio.Task(j.result())
+
+    _, pending = await asyncio.wait([result_call], timeout=0.1)
+    assert pending == {result_call}
+
+    await worker.main()
+
+    with pytest.raises(ResultNotFound):
+        # Job has completed and did not store any result
+        await asyncio.wait_for(result_call, timeout=5)
+
+
 async def test_enqueue_job(arq_redis: ArqRedis, worker, queue_name=default_queue_name):
     async def foobar(ctx, *args, **kwargs):
         return 42

tests/test_worker.py

@@ -799,6 +799,11 @@ async def wait_and_abort(job, delay=0.1):
     assert worker.job_tasks == {}
 
 
+async def test_abort_job_which_is_not_in_queue(arq_redis: ArqRedis):
+    job = Job(job_id='testing', redis=arq_redis)
+    assert await job.abort() is False
+
+
 async def test_abort_job_before(arq_redis: ArqRedis, worker, caplog, loop):
     async def longfunc(ctx):
         await asyncio.sleep(3600)