🏗️ Architect: Decouple Generic and EDC Endpoint Logic

imednet/core/endpoint/base.py

@@ -9,20 +9,22 @@
 
 from imednet.core.context import Context
 from imednet.core.endpoint.abc import EndpointABC
+from imednet.core.endpoint.edc_mixin import EdcEndpointMixin
 from imednet.core.protocols import AsyncRequestorProtocol, RequestorProtocol
 from imednet.models.json_base import JsonModel
 
 T = TypeVar("T", bound=JsonModel)
 
 
-class BaseEndpoint(EndpointABC[T]):
+class GenericEndpoint(EndpointABC[T]):
     """
-    Shared base for endpoint wrappers.
+    Generic base for endpoint wrappers.
 
-    Handles context injection and filtering.
+    Handles context injection and basic path building.
+    Does NOT include EDC-specific logic.
     """
 
-    BASE_PATH = "/api/v1/edc/studies"
+    BASE_PATH = ""
 
     def __init__(
         self,
@@ -43,15 +45,21 @@ def __init__(
                 setattr(self, cache_name, None)
 
     def _auto_filter(self, filters: Dict[str, Any]) -> Dict[str, Any]:
-        # inject default studyKey if missing
-        if "studyKey" not in filters and self._ctx.default_study_key:
-            filters["studyKey"] = self._ctx.default_study_key
+        """Pass-through for filters in generic endpoints."""
         return filters
 
     def _build_path(self, *segments: Any) -> str:
-        """Return an API path joined with :data:`BASE_PATH`."""
+        """
+        Return an API path joined with :data:`BASE_PATH`.
 
-        parts = [self.BASE_PATH.strip("/")]
+        Args:
+            *segments: URL path segments to append.
+
+        Returns:
+            The full API path string.
+        """
+        base = self.BASE_PATH.strip("/")
+        parts = [base] if base else []
         for seg in segments:
             text = str(seg).strip("/")
             if text:
@@ -64,3 +72,14 @@ def _require_async_client(self) -> AsyncRequestorProtocol:
         if self._async_client is None:
             raise RuntimeError("Async client not configured")
         return self._async_client
+
+
+class BaseEndpoint(EdcEndpointMixin, GenericEndpoint[T]):
+    """
+    Shared base for endpoint wrappers (Legacy).
+
+    Includes EDC-specific logic for backward compatibility.
+    New endpoints should use GenericEndpoint or explicit Edc* mixins.
+    """
+
+    pass

imednet/core/endpoint/edc_mixin.py

@@ -0,0 +1,55 @@
+"""Mixin for EDC-specific endpoint logic."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Dict
+from urllib.parse import quote
+
+if TYPE_CHECKING:
+    from imednet.core.context import Context
+
+
+class EdcEndpointMixin:
+    """
+    Mixin providing EDC-specific logic for endpoints.
+
+    This includes the base path for EDC resources and automatic injection
+    of the default study key into filters.
+    """
+
+    BASE_PATH = "/api/v1/edc/studies"
+
+    if TYPE_CHECKING:
+        _ctx: Context
+
+    def _auto_filter(self, filters: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Inject default studyKey if missing.
+
+        Args:
+            filters: The current dictionary of filters.
+
+        Returns:
+            The filters dictionary with studyKey injected if applicable.
+        """
+        if "studyKey" not in filters and self._ctx.default_study_key:
+            filters["studyKey"] = self._ctx.default_study_key
+        return filters
+
+    def _build_path(self, *segments: Any) -> str:
+        """
+        Return an API path joined with :data:`BASE_PATH`.
+
+        Args:
+            *segments: URL path segments to append.
+
+        Returns:
+            The full API path string.
+        """
+        parts = [self.BASE_PATH.strip("/")]
+        for seg in segments:
+            text = str(seg).strip("/")
+            if text:
+                # Encode path segments to prevent traversal and injection
+                parts.append(quote(text, safe=""))
+        return "/" + "/".join(parts)

imednet/core/endpoint/mixins/__init__.py

@@ -2,6 +2,14 @@
 from imednet.utils.filters import build_filter_string
 
 from .bases import (
+    EdcListEndpoint,
+    EdcListGetEndpoint,
+    EdcListPathGetEndpoint,
+    EdcMetadataListGetEndpoint,
+    EdcStrictListGetEndpoint,
+    GenericListEndpoint,
+    GenericListGetEndpoint,
+    GenericListPathGetEndpoint,
     ListEndpoint,
     ListGetEndpoint,
     ListGetEndpointMixin,
@@ -20,7 +28,15 @@
     "AsyncPaginator",
     "CacheMixin",
     "CreateEndpointMixin",
+    "EdcListEndpoint",
+    "EdcListGetEndpoint",
+    "EdcListPathGetEndpoint",
+    "EdcMetadataListGetEndpoint",
+    "EdcStrictListGetEndpoint",
     "FilterGetEndpointMixin",
+    "GenericListEndpoint",
+    "GenericListGetEndpoint",
+    "GenericListPathGetEndpoint",
     "ListEndpoint",
     "ListEndpointMixin",
     "ListGetEndpoint",

imednet/core/endpoint/mixins/bases.py

@@ -2,10 +2,11 @@
 
 from typing import Any, Awaitable, List, Optional, cast
 
+from imednet.core.endpoint.base import GenericEndpoint
+from imednet.core.endpoint.edc_mixin import EdcEndpointMixin
 from imednet.core.paginator import AsyncPaginator, Paginator
 from imednet.core.protocols import AsyncRequestorProtocol, RequestorProtocol
 
-from ..base import BaseEndpoint
 from .get import FilterGetEndpointMixin, PathGetEndpointMixin
 from .list import ListEndpointMixin
 from .parsing import T
@@ -17,8 +18,8 @@ class ListGetEndpointMixin(ListEndpointMixin[T], FilterGetEndpointMixin[T]):
     pass
 
 
-class ListEndpoint(BaseEndpoint, ListEndpointMixin[T]):
-    """Endpoint base class implementing ``list`` helpers."""
+class GenericListEndpoint(GenericEndpoint[T], ListEndpointMixin[T]):
+    """Generic endpoint implementing ``list`` helpers."""
 
     PAGINATOR_CLS: type[Paginator] = Paginator
     ASYNC_PAGINATOR_CLS: type[AsyncPaginator] = AsyncPaginator
@@ -43,8 +44,20 @@ async def async_list(self, study_key: Optional[str] = None, **filters: Any) -> L
         )
 
 
-class ListGetEndpoint(ListEndpoint[T], FilterGetEndpointMixin[T]):
-    """Endpoint base class implementing ``list`` and ``get`` helpers."""
+class EdcListEndpoint(EdcEndpointMixin, GenericListEndpoint[T]):
+    """EDC-specific list endpoint."""
+
+    pass
+
+
+class ListEndpoint(EdcListEndpoint[T]):
+    """Endpoint base class implementing ``list`` helpers (Legacy)."""
+
+    pass
+
+
+class GenericListGetEndpoint(GenericListEndpoint[T], FilterGetEndpointMixin[T]):
+    """Generic endpoint implementing ``list`` and ``get`` helpers."""
 
     def _get_common(
         self,
@@ -65,8 +78,20 @@ async def async_get(self, study_key: Optional[str], item_id: Any) -> T:
         )
 
 
-class ListPathGetEndpoint(ListEndpoint[T], PathGetEndpointMixin[T]):
-    """Endpoint base class implementing ``list`` and ``get`` (via path) helpers."""
+class EdcListGetEndpoint(EdcEndpointMixin, GenericListGetEndpoint[T]):
+    """EDC-specific list/get endpoint."""
+
+    pass
+
+
+class ListGetEndpoint(EdcListGetEndpoint[T]):
+    """Endpoint base class implementing ``list`` and ``get`` helpers (Legacy)."""
+
+    pass
+
+
+class GenericListPathGetEndpoint(GenericListEndpoint[T], PathGetEndpointMixin[T]):
+    """Generic endpoint implementing ``list`` and ``get`` (via path) helpers."""
 
     def get(self, study_key: Optional[str], item_id: Any) -> T:
         return cast(T, self._get_impl_path(self._client, study_key=study_key, item_id=item_id))
@@ -79,9 +104,21 @@ async def async_get(self, study_key: Optional[str], item_id: Any) -> T:
         )
 
 
-class StrictListGetEndpoint(ListGetEndpoint[T]):
+class EdcListPathGetEndpoint(EdcEndpointMixin, GenericListPathGetEndpoint[T]):
+    """EDC-specific list/path-get endpoint."""
+
+    pass
+
+
+class ListPathGetEndpoint(EdcListPathGetEndpoint[T]):
+    """Endpoint base class implementing ``list`` and ``get`` (via path) helpers (Legacy)."""
+
+    pass
+
+
+class EdcStrictListGetEndpoint(EdcListGetEndpoint[T]):
     """
-    Endpoint base class enforcing strict study key requirements.
+    Endpoint base class enforcing strict study key requirements (EDC).
 
     Populates study key from filters and raises KeyError if missing.
     """
@@ -90,11 +127,23 @@ class StrictListGetEndpoint(ListGetEndpoint[T]):
     _missing_study_exception = KeyError
 
 
-class MetadataListGetEndpoint(StrictListGetEndpoint[T]):
+class StrictListGetEndpoint(EdcStrictListGetEndpoint[T]):
+    """Endpoint base class enforcing strict study key requirements (Legacy)."""
+
+    pass
+
+
+class EdcMetadataListGetEndpoint(EdcStrictListGetEndpoint[T]):
     """
-    Endpoint base class for metadata resources.
+    Endpoint base class for metadata resources (EDC).
 
     Inherits strict study key requirements and sets a larger default page size.
     """
 
     PAGE_SIZE = 500
+
+
+class MetadataListGetEndpoint(EdcMetadataListGetEndpoint[T]):
+    """Endpoint base class for metadata resources (Legacy)."""
+
+    pass

imednet/endpoints/codings.py

@@ -1,10 +1,10 @@
 """Endpoint for managing codings (medical coding) in a study."""
 
-from imednet.core.endpoint.mixins import StrictListGetEndpoint
+from imednet.core.endpoint.mixins import EdcStrictListGetEndpoint
 from imednet.models.codings import Coding
 
 
-class CodingsEndpoint(StrictListGetEndpoint[Coding]):
+class CodingsEndpoint(EdcStrictListGetEndpoint[Coding]):
     """
     API endpoint for interacting with codings (medical coding) in an iMedNet study.
 

imednet/endpoints/forms.py

@@ -1,10 +1,10 @@
 """Endpoint for managing forms (eCRFs) in a study."""
 
-from imednet.core.endpoint.mixins import MetadataListGetEndpoint
+from imednet.core.endpoint.mixins import EdcMetadataListGetEndpoint
 from imednet.models.forms import Form
 
 
-class FormsEndpoint(MetadataListGetEndpoint[Form]):
+class FormsEndpoint(EdcMetadataListGetEndpoint[Form]):
     """
     API endpoint for interacting with forms (eCRFs) in an iMedNet study.
 

imednet/endpoints/intervals.py

@@ -1,10 +1,10 @@
 """Endpoint for managing intervals (visit definitions) in a study."""
 
-from imednet.core.endpoint.mixins import MetadataListGetEndpoint
+from imednet.core.endpoint.mixins import EdcMetadataListGetEndpoint
 from imednet.models.intervals import Interval
 
 
-class IntervalsEndpoint(MetadataListGetEndpoint[Interval]):
+class IntervalsEndpoint(EdcMetadataListGetEndpoint[Interval]):
     """
     API endpoint for interacting with intervals (visit definitions) in an iMedNet study.
 

imednet/endpoints/jobs.py

@@ -2,12 +2,12 @@
 
 from typing import Any, Optional
 
-from imednet.core.endpoint.mixins import ListPathGetEndpoint
+from imednet.core.endpoint.mixins import EdcListPathGetEndpoint
 from imednet.core.paginator import AsyncJsonListPaginator, JsonListPaginator
 from imednet.models.jobs import JobStatus
 
 
-class JobsEndpoint(ListPathGetEndpoint[JobStatus]):
+class JobsEndpoint(EdcListPathGetEndpoint[JobStatus]):
     """
     API endpoint for retrieving status and details of jobs in an iMedNet study.
 

imednet/endpoints/queries.py

@@ -1,10 +1,10 @@
 """Endpoint for managing queries (dialogue/questions) in a study."""
 
-from imednet.core.endpoint.mixins import ListGetEndpoint
+from imednet.core.endpoint.mixins import EdcListGetEndpoint
 from imednet.models.queries import Query
 
 
-class QueriesEndpoint(ListGetEndpoint[Query]):
+class QueriesEndpoint(EdcListGetEndpoint[Query]):
     """
     API endpoint for interacting with queries (dialogue/questions) in an iMedNet study.
 

imednet/endpoints/record_revisions.py

@@ -1,10 +1,10 @@
 """Endpoint for retrieving record revision history in a study."""
 
-from imednet.core.endpoint.mixins import ListGetEndpoint
+from imednet.core.endpoint.mixins import EdcListGetEndpoint
 from imednet.models.record_revisions import RecordRevision
 
 
-class RecordRevisionsEndpoint(ListGetEndpoint[RecordRevision]):
+class RecordRevisionsEndpoint(EdcListGetEndpoint[RecordRevision]):
     """
     API endpoint for accessing record revision history in an iMedNet study.
 

imednet/endpoints/records.py

@@ -3,13 +3,13 @@
 from typing import Any, Dict, List, Optional, Union
 
 from imednet.constants import HEADER_EMAIL_NOTIFY
-from imednet.core.endpoint.mixins import CreateEndpointMixin, ListGetEndpoint
+from imednet.core.endpoint.mixins import CreateEndpointMixin, EdcListGetEndpoint
 from imednet.models.jobs import Job
 from imednet.models.records import Record
 from imednet.validation.cache import SchemaCache, validate_record_data
 
 
-class RecordsEndpoint(ListGetEndpoint[Record], CreateEndpointMixin[Job]):
+class RecordsEndpoint(EdcListGetEndpoint[Record], CreateEndpointMixin[Job]):
     """
     API endpoint for interacting with records (eCRF instances) in an iMedNet study.
 

imednet/endpoints/registry.py

@@ -9,7 +9,7 @@
 
 from typing import Dict, Type
 
-from imednet.core.endpoint.base import BaseEndpoint
+from imednet.core.endpoint.base import GenericEndpoint
 from imednet.endpoints.codings import CodingsEndpoint
 from imednet.endpoints.forms import FormsEndpoint
 from imednet.endpoints.intervals import IntervalsEndpoint
@@ -24,7 +24,7 @@
 from imednet.endpoints.variables import VariablesEndpoint
 from imednet.endpoints.visits import VisitsEndpoint
 
-ENDPOINT_REGISTRY: Dict[str, Type[BaseEndpoint]] = {
+ENDPOINT_REGISTRY: Dict[str, Type[GenericEndpoint]] = {
     "codings": CodingsEndpoint,
     "forms": FormsEndpoint,
     "intervals": IntervalsEndpoint,