"""
Base endpoint mix-in for all API resource endpoints.
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional, TypeVar, Union
from urllib.parse import quote
from imednet.core.context import Context
from imednet.core.endpoint.abc import EndpointABC
from imednet.core.endpoint.mixins import FilterGetEndpointMixin, ListEndpointMixin
from imednet.core.protocols import AsyncRequestorProtocol, RequestorProtocol
from imednet.models.json_base import JsonModel
T = TypeVar("T", bound=JsonModel)
[docs]class GenericEndpoint(EndpointABC[T]):
"""
Generic base for endpoint wrappers.
Handles context injection and basic path building.
Does NOT include EDC-specific logic.
"""
BASE_PATH = ""
_client: RequestorProtocol
_async_client: Optional[AsyncRequestorProtocol]
_cache: Optional[Union[List[T], Dict[str, List[T]]]] = None
def __init__(
self,
client: RequestorProtocol,
ctx: Context,
async_client: Optional[AsyncRequestorProtocol] = None,
) -> None:
self._client = client
self._async_client = async_client
self._ctx = ctx
# Initialize cache if configured
if self._enable_cache:
if self.requires_study_key:
self._cache = {}
else:
self._cache = None
def _auto_filter(self, filters: Dict[str, Any]) -> Dict[str, Any]:
"""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`.
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:
# Encode path segments to prevent traversal and injection
parts.append(quote(text, safe=""))
return "/" + "/".join(parts)
def _require_sync_client(self) -> RequestorProtocol:
"""Return the configured sync client."""
return self._client
def _require_async_client(self) -> AsyncRequestorProtocol:
"""Return the configured async client or raise if missing."""
if self._async_client is None:
raise RuntimeError("Async client not configured")
return self._async_client
[docs]class GenericListGetEndpoint(
GenericEndpoint[T],
ListEndpointMixin[T],
FilterGetEndpointMixin[T],
):
"""
Generic base for endpoints that provide list and get-by-filter functionality.
Combines GenericEndpoint with ListEndpointMixin and FilterGetEndpointMixin
to provide standard CRUD read operations.
"""
pass