🛠️ Refactor: Endpoints - Strategy Pattern & DRY List Logic

imednet/core/endpoint/mixins/bases.py

@@ -2,6 +2,7 @@
 
 from imednet.core.endpoint.base import GenericEndpoint
 from imednet.core.endpoint.edc_mixin import EdcEndpointMixin
+from imednet.core.endpoint.strategies import PopStudyKeyStrategy
 from imednet.core.paginator import AsyncPaginator, Paginator  # noqa: F401
 
 from .get import FilterGetEndpointMixin, PathGetEndpointMixin
@@ -76,7 +77,7 @@ class EdcStrictListGetEndpoint(EdcListGetEndpoint[T]):
     Populates study key from filters and raises KeyError if missing.
     """
 
-    _pop_study_filter = True
+    STUDY_KEY_STRATEGY = PopStudyKeyStrategy
     _missing_study_exception = KeyError
 
 

imednet/core/endpoint/mixins/list.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Callable, Dict, Iterable, List, Optional, cast
+from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union, cast
 
 from imednet.constants import DEFAULT_PAGE_SIZE
 from imednet.core.endpoint.abc import EndpointABC
@@ -128,6 +128,19 @@ def _prepare_list_request(
             cache=cache,
         )
 
+    def _create_paginator_and_parser(
+        self,
+        client: Union[RequestorProtocol, AsyncRequestorProtocol],
+        paginator_cls: Union[type[Paginator], type[AsyncPaginator]],
+        state: ListRequestState[T],
+    ) -> Tuple[Union[Paginator, AsyncPaginator], Callable[[Any], T]]:
+        """Create paginator and resolve parser."""
+        paginator = paginator_cls(  # type: ignore[operator]
+            client, state.path, params=state.params, page_size=self.PAGE_SIZE
+        )
+        parse_func = self._resolve_parse_func()
+        return paginator, parse_func
+
     def _list_sync(
         self,
         client: RequestorProtocol,
@@ -143,11 +156,10 @@ def _list_sync(
         if state.cached_result is not None:
             return state.cached_result
 
-        paginator = paginator_cls(client, state.path, params=state.params, page_size=self.PAGE_SIZE)
-        parse_func = self._resolve_parse_func()
+        paginator, parse_func = self._create_paginator_and_parser(client, paginator_cls, state)
 
         return self._execute_sync_list(
-            paginator,
+            cast(Paginator, paginator),
             parse_func,
             state.study,
             state.has_filters,
@@ -169,11 +181,10 @@ async def _list_async(
         if state.cached_result is not None:
             return state.cached_result
 
-        paginator = paginator_cls(client, state.path, params=state.params, page_size=self.PAGE_SIZE)
-        parse_func = self._resolve_parse_func()
+        paginator, parse_func = self._create_paginator_and_parser(client, paginator_cls, state)
 
         return await self._execute_async_list(
-            paginator,
+            cast(AsyncPaginator, paginator),
             parse_func,
             state.study,
             state.has_filters,

imednet/core/endpoint/mixins/params.py

@@ -1,8 +1,12 @@
 from __future__ import annotations
 
-from typing import Any, Dict, Optional, cast
+from typing import Any, Dict, Optional, Type, cast
 
-from imednet.core.endpoint.strategies import DefaultParamProcessor
+from imednet.core.endpoint.strategies import (
+    DefaultParamProcessor,
+    KeepStudyKeyStrategy,
+    StudyKeyStrategy,
+)
 from imednet.core.endpoint.structs import ParamState
 from imednet.core.protocols import ParamProcessor
 from imednet.utils.filters import build_filter_string
@@ -14,10 +18,28 @@ class ParamMixin:
     """Mixin for handling endpoint parameters and filters."""
 
     requires_study_key: bool = True
-    _pop_study_filter: bool = False
     _missing_study_exception: type[Exception] = ValueError
 
     PARAM_PROCESSOR_CLS: type[ParamProcessor] = DefaultParamProcessor
+    STUDY_KEY_STRATEGY: Type[StudyKeyStrategy] = KeepStudyKeyStrategy
+
+    # Backward compatibility for subclasses that haven't migrated
+    _pop_study_filter: bool = False
+
+    def _resolve_study_strategy(self) -> StudyKeyStrategy:
+        """Resolve the study key strategy."""
+        # If the class has overridden STUDY_KEY_STRATEGY, use it.
+        if self.STUDY_KEY_STRATEGY is not KeepStudyKeyStrategy:
+             return self.STUDY_KEY_STRATEGY(self.requires_study_key, self._missing_study_exception)
+
+        # Fallback to checking legacy flag if strategy is default
+        # But for cleaner refactor, we should assume subclasses are updated or we update them.
+        # However, to be safe during refactor:
+        if self._pop_study_filter:
+            from imednet.core.endpoint.strategies import PopStudyKeyStrategy
+            return PopStudyKeyStrategy(self.requires_study_key, self._missing_study_exception)
+
+        return self.STUDY_KEY_STRATEGY(self.requires_study_key, self._missing_study_exception)
 
     def _resolve_params(
         self,
@@ -41,27 +63,14 @@ def _resolve_params(
         if study_key:
             filters["studyKey"] = study_key
 
-        study: Optional[str] = None
-        if self.requires_study_key:
-            if self._pop_study_filter:
-                try:
-                    study = filters.pop("studyKey")
-                except KeyError as exc:
-                    raise self._missing_study_exception(
-                        "Study key must be provided or set in the context"
-                    ) from exc
-            else:
-                study = filters.get("studyKey")
-                if not study:
-                    raise ValueError("Study key must be provided or set in the context")
-        else:
-            study = filters.get("studyKey")
-
-        other_filters = {k: v for k, v in filters.items() if k != "studyKey"}
+        strategy = self._resolve_study_strategy()
+        study, query_filters = strategy.extract(filters)
+
+        other_filters = {k: v for k, v in query_filters.items() if k != "studyKey"}
 
         params: Dict[str, Any] = {}
-        if filters:
-            params["filter"] = build_filter_string(filters)
+        if query_filters:
+            params["filter"] = build_filter_string(query_filters)
         if extra_params:
             params.update(extra_params)
 

imednet/core/endpoint/strategies.py

@@ -5,7 +5,8 @@
 to customize how filters are processed and special parameters are extracted.
 """
 
-from typing import Any, Dict, Tuple
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List, Optional, Tuple, Type
 
 from imednet.core.protocols import ParamProcessor
 
@@ -28,3 +29,104 @@ def process_filters(self, filters: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict
             A tuple of (copy of filters, empty dict).
         """
         return filters.copy(), {}
+
+
+@dataclass
+class ParamRule:
+    """Rule for mapping a filter key to a parameter."""
+
+    source: str
+    target: str
+    transform: Callable[[Any], Any] = field(default_factory=lambda: lambda x: x)
+    default: Any = None
+    skip_none: bool = True
+    skip_falsey: bool = False
+
+
+class MappingParamProcessor(ParamProcessor):
+    """
+    Declarative parameter processor.
+
+    Iterates over defined rules to process filters, extracting special parameters.
+    """
+
+    rules: List[ParamRule] = []
+
+    def process_filters(self, filters: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
+        """
+        Process filters based on configured rules.
+
+        Args:
+            filters: The input filters dictionary.
+
+        Returns:
+            A tuple of (cleaned filters, special parameters).
+        """
+        filters = filters.copy()
+        special_params: Dict[str, Any] = {}
+
+        for rule in self.rules:
+            # Pop the source key if present, otherwise use default
+            value = filters.pop(rule.source, rule.default)
+
+            # If value is None and skip_none is True, skip
+            if value is None and rule.skip_none:
+                continue
+
+            transformed = rule.transform(value)
+
+            # If transformed value is falsey and skip_falsey is True, skip
+            if not transformed and rule.skip_falsey:
+                continue
+
+            special_params[rule.target] = transformed
+
+        return filters, special_params
+
+
+class StudyKeyStrategy:
+    """Strategy for handling study key extraction from filters."""
+
+    def __init__(self, requires_study_key: bool, missing_exception: Type[Exception] = ValueError):
+        self.requires_study_key = requires_study_key
+        self.missing_exception = missing_exception
+
+    def extract(self, filters: Dict[str, Any]) -> Tuple[Optional[str], Dict[str, Any]]:
+        """
+        Extract study key from filters.
+
+        Args:
+            filters: The filters dictionary.
+
+        Returns:
+            Tuple of (study_key, filters_for_query).
+        """
+        raise NotImplementedError
+
+
+class KeepStudyKeyStrategy(StudyKeyStrategy):
+    """Strategy that keeps the study key in filters (validation only)."""
+
+    def extract(self, filters: Dict[str, Any]) -> Tuple[Optional[str], Dict[str, Any]]:
+        filters = filters.copy()
+        study = filters.get("studyKey")
+        if not study and self.requires_study_key:
+            raise self.missing_exception("Study key must be provided or set in the context")
+        return study, filters
+
+
+class PopStudyKeyStrategy(StudyKeyStrategy):
+    """Strategy that pops the study key from filters."""
+
+    def extract(self, filters: Dict[str, Any]) -> Tuple[Optional[str], Dict[str, Any]]:
+        filters = filters.copy()
+        try:
+            study = filters.pop("studyKey")
+        except KeyError as exc:
+            if self.requires_study_key:
+                raise self.missing_exception(
+                    "Study key must be provided or set in the context"
+                ) from exc
+            study = None
+
+        return study, filters

imednet/endpoints/records.py

@@ -1,34 +1,23 @@
 """Endpoint for managing records (eCRF instances) in a study."""
 
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Union
 
 from imednet.constants import HEADER_EMAIL_NOTIFY
 from imednet.core.endpoint.mixins import CreateEndpointMixin, EdcListGetEndpoint
-from imednet.core.protocols import ParamProcessor
+from imednet.core.endpoint.strategies import (
+    KeepStudyKeyStrategy,
+    MappingParamProcessor,
+    ParamRule,
+)
 from imednet.models.jobs import Job
 from imednet.models.records import Record
 from imednet.validation.cache import SchemaCache, validate_record_data
 
 
-class RecordsParamProcessor(ParamProcessor):
+class RecordsParamProcessor(MappingParamProcessor):
     """Parameter processor for Records endpoint."""
 
-    def process_filters(self, filters: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
-        """
-        Extract 'record_data_filter' parameter.
-
-        Args:
-            filters: The filters dictionary.
-
-        Returns:
-            Tuple of (cleaned filters, special parameters).
-        """
-        filters = filters.copy()
-        record_data_filter = filters.pop("record_data_filter", None)
-        special_params = {}
-        if record_data_filter:
-            special_params["recordDataFilter"] = record_data_filter
-        return filters, special_params
+    rules = [ParamRule(source="record_data_filter", target="recordDataFilter")]
 
 
 class RecordsEndpoint(EdcListGetEndpoint[Record], CreateEndpointMixin[Job]):
@@ -41,7 +30,7 @@ class RecordsEndpoint(EdcListGetEndpoint[Record], CreateEndpointMixin[Job]):
     PATH = "records"
     MODEL = Record
     _id_param = "recordId"
-    _pop_study_filter = False
+    STUDY_KEY_STRATEGY = KeepStudyKeyStrategy
     PARAM_PROCESSOR_CLS = RecordsParamProcessor
 
     def _prepare_create_request(

imednet/endpoints/users.py

@@ -1,29 +1,25 @@
 """Endpoint for managing users in a study."""
 
-from typing import Any, Dict, Tuple
-
 from imednet.core.endpoint.mixins import EdcListGetEndpoint
-from imednet.core.protocols import ParamProcessor
+from imednet.core.endpoint.strategies import (
+    MappingParamProcessor,
+    ParamRule,
+    PopStudyKeyStrategy,
+)
 from imednet.models.users import User
 
 
-class UsersParamProcessor(ParamProcessor):
+class UsersParamProcessor(MappingParamProcessor):
     """Parameter processor for Users endpoint."""
 
-    def process_filters(self, filters: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
-        """
-        Extract 'include_inactive' parameter.
-
-        Args:
-            filters: The filters dictionary.
-
-        Returns:
-            Tuple of (cleaned filters, special parameters).
-        """
-        filters = filters.copy()
-        include_inactive = filters.pop("include_inactive", False)
-        special_params = {"includeInactive": str(include_inactive).lower()}
-        return filters, special_params
+    rules = [
+        ParamRule(
+            source="include_inactive",
+            target="includeInactive",
+            default=False,
+            transform=lambda x: str(x).lower(),
+        )
+    ]
 
 
 class UsersEndpoint(EdcListGetEndpoint[User]):
@@ -36,5 +32,5 @@ class UsersEndpoint(EdcListGetEndpoint[User]):
     PATH = "users"
     MODEL = User
     _id_param = "userId"
-    _pop_study_filter = True
+    STUDY_KEY_STRATEGY = PopStudyKeyStrategy
     PARAM_PROCESSOR_CLS = UsersParamProcessor