Refactor: Implement Strategy pattern for params and study key handling

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,8 +77,7 @@ class EdcStrictListGetEndpoint(EdcListGetEndpoint[T]):
     Populates study key from filters and raises KeyError if missing.
     """
 
-    _pop_study_filter = True
-    _missing_study_exception = KeyError
+    STUDY_KEY_STRATEGY = PopStudyKeyStrategy(KeyError)
 
 
 class StrictListGetEndpoint(EdcStrictListGetEndpoint[T]):

imednet/core/endpoint/mixins/caching.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, cast
+
+from ..protocols import EndpointProtocol
 
 
 class CacheMixin:
     """Mixin for handling endpoint caching."""
 
-    requires_study_key: bool = True  # Default, can be overridden
     _enable_cache: bool = False  # Default, overridden by EndpointABC/subclasses
     _cache: Any = None  # Default, overridden by GenericEndpoint
 
@@ -25,7 +26,7 @@ def _update_local_cache(
         if has_filters or not self._enable_cache:
             return
 
-        if self.requires_study_key:
+        if cast(EndpointProtocol, self).requires_study_key:
             if self._cache is not None:
                 self._cache[study] = result
         else:
@@ -41,7 +42,7 @@ def _check_cache_hit(
         if not self._enable_cache:
             return None
 
-        if self.requires_study_key:
+        if cast(EndpointProtocol, self).requires_study_key:
             # Strict check usually done before, but here we just check cache
             if cache is not None and not other_filters and not refresh and study in cache:
                 return cache[study]

imednet/core/endpoint/mixins/params.py

@@ -2,7 +2,11 @@
 
 from typing import Any, Dict, Optional, 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
@@ -13,12 +17,17 @@
 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
+    # Default strategy: Keep studyKey, raise ValueError if missing
+    STUDY_KEY_STRATEGY: StudyKeyStrategy = KeepStudyKeyStrategy(ValueError)
 
+    PARAM_PROCESSOR: Optional[ParamProcessor] = None
     PARAM_PROCESSOR_CLS: type[ParamProcessor] = DefaultParamProcessor
 
+    @property
+    def requires_study_key(self) -> bool:
+        """Whether this endpoint requires a study key."""
+        return self.STUDY_KEY_STRATEGY.is_required
+
     def _resolve_params(
         self,
         study_key: Optional[str],
@@ -30,7 +39,7 @@ def _resolve_params(
         filters = cast(EndpointProtocol, self)._auto_filter(filters)
 
         # Use the configured parameter processor strategy
-        processor = self.PARAM_PROCESSOR_CLS()
+        processor = self.PARAM_PROCESSOR or self.PARAM_PROCESSOR_CLS()
         filters, special_params = processor.process_filters(filters)
 
         if special_params:
@@ -41,21 +50,7 @@ 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")
+        study, filters = self.STUDY_KEY_STRATEGY.extract_study_key(filters)
 
         other_filters = {k: v for k, v in filters.items() if k != "studyKey"}
 

imednet/core/endpoint/strategies.py

@@ -5,11 +5,119 @@
 to customize how filters are processed and special parameters are extracted.
 """
 
-from typing import Any, Dict, Tuple
+from dataclasses import dataclass
+from typing import Any, Callable, Dict, Optional, Protocol, Tuple, Type
 
 from imednet.core.protocols import ParamProcessor
 
 
+class StudyKeyStrategy(Protocol):
+    """Strategy for handling study key extraction and validation."""
+
+    @property
+    def is_required(self) -> bool:
+        """Whether the study key is required."""
+        ...
+
+    def extract_study_key(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, modified_filters).
+        """
+        ...
+
+
+class PopStudyKeyStrategy:
+    """Strategy that removes studyKey from filters."""
+
+    def __init__(self, exception_cls: Type[Exception] = ValueError) -> None:
+        self.exception_cls = exception_cls
+
+    @property
+    def is_required(self) -> bool:
+        return True
+
+    def extract_study_key(self, filters: Dict[str, Any]) -> Tuple[Optional[str], Dict[str, Any]]:
+        filters = filters.copy()
+        try:
+            study_key = filters.pop("studyKey")
+        except KeyError as exc:
+            raise self.exception_cls("Study key must be provided or set in the context") from exc
+        return study_key, filters
+
+
+class KeepStudyKeyStrategy:
+    """Strategy that keeps studyKey in filters."""
+
+    def __init__(self, exception_cls: Type[Exception] = ValueError) -> None:
+        self.exception_cls = exception_cls
+
+    @property
+    def is_required(self) -> bool:
+        return True
+
+    def extract_study_key(self, filters: Dict[str, Any]) -> Tuple[Optional[str], Dict[str, Any]]:
+        filters = filters.copy()
+        study_key = filters.get("studyKey")
+        if not study_key:
+            raise self.exception_cls("Study key must be provided or set in the context")
+        return study_key, filters
+
+
+class OptionalStudyKeyStrategy:
+    """Strategy where studyKey is optional and kept in filters."""
+
+    @property
+    def is_required(self) -> bool:
+        return False
+
+    def extract_study_key(self, filters: Dict[str, Any]) -> Tuple[Optional[str], Dict[str, Any]]:
+        return filters.get("studyKey"), filters.copy()
+
+
+@dataclass
+class ParamRule:
+    """Rule for mapping a filter parameter."""
+
+    input_key: str
+    output_key: str
+    default: Any = None
+    transform: Optional[Callable[[Any], Any]] = None
+    skip_none: bool = True
+    skip_falsey: bool = False
+
+
+class MappingParamProcessor(ParamProcessor):
+    """Parameter processor using declarative mapping rules."""
+
+    def __init__(self, rules: list[ParamRule]) -> None:
+        self.rules = rules
+
+    def process_filters(self, filters: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
+        filters = filters.copy()
+        special_params = {}
+
+        for rule in self.rules:
+            value = filters.pop(rule.input_key, rule.default)
+
+            if rule.skip_none and value is None:
+                continue
+            if rule.skip_falsey and not value:
+                continue
+
+            if rule.transform and value is not None:
+                value = rule.transform(value)
+
+            special_params[rule.output_key] = value
+
+        return filters, special_params
+
+
 class DefaultParamProcessor(ParamProcessor):
     """
     Default parameter processor.

imednet/endpoints/records.py

@@ -1,34 +1,17 @@
 """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):
-    """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
+from imednet.validation.cache import SchemaCache, validate_record_entry
 
 
 class RecordsEndpoint(EdcListGetEndpoint[Record], CreateEndpointMixin[Job]):
@@ -41,8 +24,17 @@ class RecordsEndpoint(EdcListGetEndpoint[Record], CreateEndpointMixin[Job]):
     PATH = "records"
     MODEL = Record
     _id_param = "recordId"
-    _pop_study_filter = False
-    PARAM_PROCESSOR_CLS = RecordsParamProcessor
+
+    STUDY_KEY_STRATEGY = KeepStudyKeyStrategy()
+
+    PARAM_PROCESSOR = MappingParamProcessor(
+        [
+            ParamRule(
+                input_key="record_data_filter",
+                output_key="recordDataFilter",
+            )
+        ]
+    )
 
     def _prepare_create_request(
         self,
@@ -51,30 +43,14 @@ def _prepare_create_request(
         email_notify: Union[bool, str, None],
         schema: Optional[SchemaCache],
     ) -> tuple[str, Dict[str, str]]:
-        self._validate_records_if_schema_present(schema, records_data)
+        if schema is not None:
+            for rec in records_data:
+                validate_record_entry(schema, rec)
+
         headers = self._build_headers(email_notify)
         path = self._build_path(study_key, self.PATH)
         return path, headers
 
-    def _validate_records_if_schema_present(
-        self, schema: Optional[SchemaCache], records_data: List[Dict[str, Any]]
-    ) -> None:
-        """
-        Validate records against schema if provided.
-
-        Args:
-            schema: Optional schema cache for validation
-            records_data: List of record data to validate
-        """
-        if schema is not None:
-            for rec in records_data:
-                fk = rec.get("formKey") or rec.get("form_key")
-                if not fk:
-                    fid = rec.get("formId") or rec.get("form_id") or 0
-                    fk = schema.form_key_from_id(fid)
-                if fk:
-                    validate_record_data(schema, fk, rec.get("data", {}))
-
     def _build_headers(self, email_notify: Union[bool, str, None]) -> Dict[str, str]:
         """
         Build headers for record creation request.

imednet/endpoints/studies.py

@@ -1,6 +1,7 @@
 """Endpoint for managing studies in the iMedNet system."""
 
 from imednet.core.endpoint.mixins import EdcListGetEndpoint
+from imednet.core.endpoint.strategies import OptionalStudyKeyStrategy
 from imednet.models.studies import Study
 
 
@@ -15,4 +16,4 @@ class StudiesEndpoint(EdcListGetEndpoint[Study]):
     MODEL = Study
     _id_param = "studyKey"
     _enable_cache = True
-    requires_study_key = False
+    STUDY_KEY_STRATEGY = OptionalStudyKeyStrategy()