From f1e95f7318612fa23e1becba2c868e10561c76d3 Mon Sep 17 00:00:00 2001 From: fderuiter <127706008+fderuiter@users.noreply.github.com> Date: Tue, 24 Feb 2026 16:41:31 +0000 Subject: [PATCH 1/2] Refactor: Implement Strategy pattern for params and study key handling Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> --- imednet/core/endpoint/mixins/bases.py | 4 +- imednet/core/endpoint/mixins/params.py | 35 ++++---- imednet/core/endpoint/strategies.py | 110 ++++++++++++++++++++++++- imednet/endpoints/records.py | 68 +++++---------- imednet/endpoints/studies.py | 3 +- imednet/endpoints/users.py | 42 ++++------ imednet/validation/cache.py | 24 ++++++ tests/unit/endpoints/test_list_get.py | 14 ++-- 8 files changed, 200 insertions(+), 100 deletions(-) diff --git a/imednet/core/endpoint/mixins/bases.py b/imednet/core/endpoint/mixins/bases.py index 7cb7dd99..4ab73890 100644 --- a/imednet/core/endpoint/mixins/bases.py +++ b/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]): diff --git a/imednet/core/endpoint/mixins/params.py b/imednet/core/endpoint/mixins/params.py index 0f2177bf..d3f4c134 100644 --- a/imednet/core/endpoint/mixins/params.py +++ b/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"} diff --git a/imednet/core/endpoint/strategies.py b/imednet/core/endpoint/strategies.py index ff807d2a..39534b4a 100644 --- a/imednet/core/endpoint/strategies.py +++ b/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. diff --git a/imednet/endpoints/records.py b/imednet/endpoints/records.py index e4cb7373..aaa1944a 100644 --- a/imednet/endpoints/records.py +++ b/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. diff --git a/imednet/endpoints/studies.py b/imednet/endpoints/studies.py index 60803ba9..74b45b73 100644 --- a/imednet/endpoints/studies.py +++ b/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() diff --git a/imednet/endpoints/users.py b/imednet/endpoints/users.py index 2b4fe5b5..846d77a0 100644 --- a/imednet/endpoints/users.py +++ b/imednet/endpoints/users.py @@ -1,31 +1,14 @@ """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): - """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 - - class UsersEndpoint(EdcListGetEndpoint[User]): """ API endpoint for interacting with users in an iMedNet study. @@ -36,5 +19,16 @@ class UsersEndpoint(EdcListGetEndpoint[User]): PATH = "users" MODEL = User _id_param = "userId" - _pop_study_filter = True - PARAM_PROCESSOR_CLS = UsersParamProcessor + + STUDY_KEY_STRATEGY = PopStudyKeyStrategy() + + PARAM_PROCESSOR = MappingParamProcessor( + [ + ParamRule( + input_key="include_inactive", + output_key="includeInactive", + default=False, + transform=lambda x: str(x).lower(), + ) + ] + ) diff --git a/imednet/validation/cache.py b/imednet/validation/cache.py index 15f6abca..15fccaf3 100644 --- a/imednet/validation/cache.py +++ b/imednet/validation/cache.py @@ -163,6 +163,30 @@ def validate_record_data( _check_type(variables[name].variable_type, value) +def validate_record_entry( + schema: BaseSchemaCache[Any], + record: Dict[str, Any], +) -> None: + """ + Validate a single record entry against the schema cache. + + Resolves form key from formKey or formId, and validates data. + + Args: + schema: The schema cache to use for validation. + record: The record dictionary containing 'formKey'/'formId' and 'data'. + + Raises: + ValidationError: If validation fails. + """ + fk = record.get("formKey") or record.get("form_key") + if not fk: + fid = record.get("formId") or record.get("form_id") or 0 + fk = schema.form_key_from_id(int(fid)) + if fk: + validate_record_data(schema, fk, record.get("data", {})) + + class SchemaValidator(_ValidatorMixin): """Validate record payloads using variable metadata from the API.""" diff --git a/tests/unit/endpoints/test_list_get.py b/tests/unit/endpoints/test_list_get.py index b51af7ac..512a293b 100644 --- a/tests/unit/endpoints/test_list_get.py +++ b/tests/unit/endpoints/test_list_get.py @@ -44,20 +44,21 @@ @pytest.mark.parametrize("cls,module,model,item_id", CASES) def test_list_and_get(dummy_client, context, paginator_factory, cls, module, model, item_id): ep = cls(dummy_client, context) + requires_study_key = ep.requires_study_key capture = paginator_factory(module, [{cls._id_param: item_id}]) - list_kwargs = {"study_key": "S1"} if getattr(cls, "requires_study_key", True) else {} + list_kwargs = {"study_key": "S1"} if requires_study_key else {} result = ep.list(**list_kwargs) expected_path = "/api/v1/edc/studies" - if getattr(cls, "requires_study_key", True): + if requires_study_key: expected_path += f"/S1/{cls.PATH}" elif cls.PATH: expected_path += f"/{cls.PATH}" assert capture["path"] == expected_path assert isinstance(result[0], model) - get_args = ("S1", item_id) if getattr(cls, "requires_study_key", True) else (None, item_id) + get_args = ("S1", item_id) if requires_study_key else (None, item_id) got = ep.get(*get_args) assert isinstance(got, model) @@ -74,19 +75,20 @@ async def test_async_list_and_get( item_id, ): ep = cls(dummy_client, context, async_client=dummy_client) + requires_study_key = ep.requires_study_key capture = async_paginator_factory(module, [{cls._id_param: item_id}]) - list_kwargs = {"study_key": "S1"} if getattr(cls, "requires_study_key", True) else {} + list_kwargs = {"study_key": "S1"} if requires_study_key else {} result = await ep.async_list(**list_kwargs) expected_path = "/api/v1/edc/studies" - if getattr(cls, "requires_study_key", True): + if requires_study_key: expected_path += f"/S1/{cls.PATH}" elif cls.PATH: expected_path += f"/{cls.PATH}" assert capture["path"] == expected_path assert isinstance(result[0], model) - get_args = ("S1", item_id) if getattr(cls, "requires_study_key", True) else (None, item_id) + get_args = ("S1", item_id) if requires_study_key else (None, item_id) got = await ep.async_get(*get_args) assert isinstance(got, model) From 3822960ce2cfab162126f2c6805c6c6ec6fcbc77 Mon Sep 17 00:00:00 2001 From: fderuiter <127706008+fderuiter@users.noreply.github.com> Date: Tue, 24 Feb 2026 16:46:59 +0000 Subject: [PATCH 2/2] Refactor: Implement Strategy pattern for params and study key handling Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> --- imednet/core/endpoint/mixins/caching.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/imednet/core/endpoint/mixins/caching.py b/imednet/core/endpoint/mixins/caching.py index d8917589..8236d96f 100644 --- a/imednet/core/endpoint/mixins/caching.py +++ b/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]