🛠️ Refactor: JobsEndpoint - Extract ListEndpoint (DRY/OCP)

imednet/core/endpoint/mixins.py

@@ -404,20 +404,35 @@ class ListGetEndpointMixin(ListEndpointMixin[T], FilterGetEndpointMixin[T]):
     pass
 
 
-class ListGetEndpoint(BaseEndpoint, ListGetEndpointMixin[T]):
-    """Endpoint base class implementing ``list`` and ``get`` helpers."""
+class ListEndpoint(BaseEndpoint, ListEndpointMixin[T]):
+    """Endpoint base class implementing ``list`` helpers."""
+
+    PAGINATOR_CLS: type[Paginator] = Paginator
+    ASYNC_PAGINATOR_CLS: type[AsyncPaginator] = AsyncPaginator
 
     def _get_context(
         self, is_async: bool
     ) -> tuple[RequestorProtocol | AsyncRequestorProtocol, type[Paginator] | type[AsyncPaginator]]:
         if is_async:
-            return self._require_async_client(), AsyncPaginator
-        return self._client, Paginator
+            return self._require_async_client(), self.ASYNC_PAGINATOR_CLS
+        return self._client, self.PAGINATOR_CLS
 
     def _list_common(self, is_async: bool, **kwargs: Any) -> List[T] | Awaitable[List[T]]:
         client, paginator = self._get_context(is_async)
         return self._list_impl(client, paginator, **kwargs)
 
+    def list(self, study_key: Optional[str] = None, **filters: Any) -> List[T]:
+        return cast(List[T], self._list_common(False, study_key=study_key, **filters))
+
+    async def async_list(self, study_key: Optional[str] = None, **filters: Any) -> List[T]:
+        return await cast(
+            Awaitable[List[T]], self._list_common(True, study_key=study_key, **filters)
+        )
+
+
+class ListGetEndpoint(ListEndpoint[T], FilterGetEndpointMixin[T]):
+    """Endpoint base class implementing ``list`` and ``get`` helpers."""
+
     def _get_common(
         self,
         is_async: bool,
@@ -428,14 +443,6 @@ def _get_common(
         client, paginator = self._get_context(is_async)
         return self._get_impl(client, paginator, study_key=study_key, item_id=item_id)
 
-    def list(self, study_key: Optional[str] = None, **filters: Any) -> List[T]:
-        return cast(List[T], self._list_common(False, study_key=study_key, **filters))
-
-    async def async_list(self, study_key: Optional[str] = None, **filters: Any) -> List[T]:
-        return await cast(
-            Awaitable[List[T]], self._list_common(True, study_key=study_key, **filters)
-        )
-
     def get(self, study_key: Optional[str], item_id: Any) -> T:
         return cast(T, self._get_common(False, study_key=study_key, item_id=item_id))
 

imednet/endpoints/jobs.py

@@ -1,14 +1,13 @@
 """Endpoint for checking job status in a study."""
 
-from typing import Any, Awaitable, List, Optional, cast
+from typing import Any, Awaitable, Optional, cast
 
-from imednet.core.endpoint.base import BaseEndpoint
-from imednet.core.endpoint.mixins import ListEndpointMixin, PathGetEndpointMixin
+from imednet.core.endpoint.mixins import ListEndpoint, PathGetEndpointMixin
 from imednet.core.paginator import AsyncJsonListPaginator, JsonListPaginator
 from imednet.models.jobs import JobStatus
 
 
-class JobsEndpoint(BaseEndpoint, ListEndpointMixin[JobStatus], PathGetEndpointMixin[JobStatus]):
+class JobsEndpoint(ListEndpoint[JobStatus], PathGetEndpointMixin[JobStatus]):
     """
     API endpoint for retrieving status and details of jobs in an iMedNet study.
 
@@ -18,6 +17,8 @@ class JobsEndpoint(BaseEndpoint, ListEndpointMixin[JobStatus], PathGetEndpointMi
 
     PATH = "jobs"
     MODEL = JobStatus
+    PAGINATOR_CLS = JsonListPaginator
+    ASYNC_PAGINATOR_CLS = AsyncJsonListPaginator
 
     def _raise_not_found(self, study_key: Optional[str], item_id: Any) -> None:
         raise ValueError(f"Job {item_id} not found in study {study_key}")
@@ -66,42 +67,3 @@ async def async_get(self, study_key: str, batch_id: str) -> JobStatus:
             Awaitable[JobStatus],
             self._get_impl_path(client, study_key=study_key, item_id=batch_id, is_async=True),
         )
-
-    def list(self, study_key: str) -> List[JobStatus]:
-        """
-        List all jobs for a specific study.
-
-        Args:
-            study_key: Study identifier
-
-        Returns:
-            List of JobStatus objects
-        """
-        return cast(
-            List[JobStatus],
-            self._list_impl(
-                self._client,
-                JsonListPaginator,
-                study_key=study_key,
-            ),
-        )
-
-    async def async_list(self, study_key: str) -> List[JobStatus]:
-        """
-        Asynchronously list all jobs for a specific study.
-
-        Args:
-            study_key: Study identifier
-
-        Returns:
-            List of JobStatus objects
-        """
-        client = self._require_async_client()
-        return await cast(
-            Awaitable[List[JobStatus]],
-            self._list_impl(
-                client,
-                AsyncJsonListPaginator,
-                study_key=study_key,
-            ),
-        )

tests/conftest.py

@@ -56,6 +56,9 @@ def __iter__(self):
         import imednet.core.endpoint.mixins as mixins_module
 
         monkeypatch.setattr(mixins_module, "Paginator", DummyPaginator)
+        # Also patch the class attribute since it's now used
+        if hasattr(mixins_module, "ListEndpoint"):
+            monkeypatch.setattr(mixins_module.ListEndpoint, "PAGINATOR_CLS", DummyPaginator)
         return captured
 
     return factory
@@ -82,6 +85,9 @@ async def __aiter__(self):
         import imednet.core.endpoint.mixins as mixins_module
 
         monkeypatch.setattr(mixins_module, "AsyncPaginator", DummyPaginator)
+        # Also patch the class attribute since it's now used
+        if hasattr(mixins_module, "ListEndpoint"):
+            monkeypatch.setattr(mixins_module.ListEndpoint, "ASYNC_PAGINATOR_CLS", DummyPaginator)
         return captured
 
     return factory