From 2b4a68120c4ac84de91a6c2d7b002340c52d9ab3 Mon Sep 17 00:00:00 2001 From: fderuiter <127706008+fderuiter@users.noreply.github.com> Date: Tue, 24 Feb 2026 15:33:03 +0000 Subject: [PATCH] Refactor Endpoints: Introduce Strategy Pattern for Params and DRY List Logic - **Strategies:** Introduced `MappingParamProcessor` and `ParamRule` for declarative parameter processing, and `StudyKeyStrategy` (`Keep`/`Pop`) for handling study key extraction. - **Refactoring:** Updated `ParamMixin` to use `StudyKeyStrategy`, replacing `_pop_study_filter` boolean logic. - **Cleanup:** Refactored `UsersParamProcessor` and `RecordsParamProcessor` to use `MappingParamProcessor`. Updated `UsersEndpoint` and `RecordsEndpoint` configuration. - **DRY:** Extracted `_create_paginator_and_parser` in `ListEndpointMixin` to unify setup logic for sync and async list operations. - **Safety:** Maintained backward compatibility for legacy endpoints not explicitly updated. Verified with full test suite passing. Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> --- imednet/core/endpoint/mixins/bases.py | 3 +- imednet/core/endpoint/mixins/list.py | 25 ++++-- imednet/core/endpoint/mixins/params.py | 53 +++++++------ imednet/core/endpoint/strategies.py | 104 ++++++++++++++++++++++++- imednet/endpoints/records.py | 29 +++---- imednet/endpoints/users.py | 34 ++++---- 6 files changed, 178 insertions(+), 70 deletions(-) diff --git a/imednet/core/endpoint/mixins/bases.py b/imednet/core/endpoint/mixins/bases.py index 7cb7dd99..ebf4f348 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,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 diff --git a/imednet/core/endpoint/mixins/list.py b/imednet/core/endpoint/mixins/list.py index f391a13e..c62b5bc0 100644 --- a/imednet/core/endpoint/mixins/list.py +++ b/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, diff --git a/imednet/core/endpoint/mixins/params.py b/imednet/core/endpoint/mixins/params.py index 0f2177bf..4cc8c9fe 100644 --- a/imednet/core/endpoint/mixins/params.py +++ b/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) diff --git a/imednet/core/endpoint/strategies.py b/imednet/core/endpoint/strategies.py index ff807d2a..3279dc86 100644 --- a/imednet/core/endpoint/strategies.py +++ b/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 diff --git a/imednet/endpoints/records.py b/imednet/endpoints/records.py index e4cb7373..d0a030c3 100644 --- a/imednet/endpoints/records.py +++ b/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( diff --git a/imednet/endpoints/users.py b/imednet/endpoints/users.py index 2b4fe5b5..9f1962f2 100644 --- a/imednet/endpoints/users.py +++ b/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