Fix(Arq): pass job_id to JobDef in new version

arq_admin/compat.py

@@ -0,0 +1,4 @@
+from arq import VERSION as ARQ_VERSION
+
+
+ARQ_VERSION_TUPLE = tuple(int(x) for x in ARQ_VERSION.split('.'))

arq_admin/queue.py

@@ -11,6 +11,7 @@
 from django.utils import timezone
 
 from arq_admin import settings
+from arq_admin.compat import ARQ_VERSION_TUPLE
 from arq_admin.job import JobInfo
 
 ARQ_PREFIX = 'arq:'
@@ -106,14 +107,19 @@ async def get_job_by_id(self, job_id: str) -> JobInfo:
                 unknown_function_msg = "Unknown, can't deserialize"
 
         if not base_info:
-            base_info = JobDef(
-                function=unknown_function_msg,
-                args=(),
-                kwargs={},
-                job_try=-1,
-                enqueue_time=timezone.now().replace(year=2077),
-                score=420,
-            )
+            parameters = {
+                'function': unknown_function_msg,
+                'args': (),
+                'kwargs': {},
+                'job_try': -1,
+                'enqueue_time': timezone.now().replace(year=2077),
+                'score': 420,
+            }
+            if ARQ_VERSION_TUPLE < (0, 26, 0):
+                base_info = JobDef(**parameters)
+            else:
+                parameters['job_id'] = job_id
+                base_info = JobDef(**parameters)
 
         job_info = JobInfo.from_base(base_info, job_id)
         job_info.status = await self._get_job_status(job_id)

setup.cfg

@@ -67,6 +67,7 @@ filterwarnings =
   error
   ignore::UserWarning:pytest.*:
   ignore::ResourceWarning:redis.*:
+  ignore:.*Call to deprecated close.*:DeprecationWarning
 junit_family = xunit1
 
 [coverage:run]

tests/test_queue.py

@@ -1,12 +1,13 @@
 import asyncio
-from typing import AsyncGenerator
+from dataclasses import dataclass
+from typing import AsyncGenerator, Optional
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 import pytest_asyncio
 from arq import ArqRedis
 from arq.constants import default_queue_name
-from arq.jobs import DeserializationError, Job, JobStatus
+from arq.jobs import DeserializationError, Job, JobStatus, JobDef
 from django.conf import settings
 
 from arq_admin.queue import Queue, QueueStats
@@ -104,6 +105,22 @@ async def test_deserialize_error(mocked_job_info: MagicMock, jobs_creator: JobsC
     assert job_info.function == "Unknown, can't deserialize"
 
 
+@dataclass
+class NewArqJobDef(JobDef):
+    job_id: Optional[str]
+
+
+@pytest.mark.asyncio()
+@patch('arq_admin.queue.JobDef', NewArqJobDef)
+@patch('arq_admin.queue.ARQ_VERSION_TUPLE', (0, 26, 0))
+@patch.object(Job, 'info')
+async def test_deserialize_error_in_arq_26(mocked_job_info: MagicMock, jobs_creator: JobsCreator, queue: Queue) -> None:
+    job = await jobs_creator.create_queued()
+    mocked_job_info.side_effect = DeserializationError()
+    job_info = await queue.get_job_by_id(job.job_id)
+    assert job_info.function == "Unknown, can't deserialize"
+
+
 @pytest.mark.asyncio()
 @patch.object(Job, 'abort')
 @pytest.mark.parametrize('success', [True, False])