Sandbox Client

Simplified client for testing healthcare services with various data sources.

This class provides an intuitive interface for: - Loading test datasets (MIMIC-on-FHIR, Synthea) - Generating synthetic FHIR data - Sending requests to healthcare services - Managing request/response lifecycle

Examples:

Load from dataset registry:

>>> client = SandboxClient(
...     url="http://localhost:8000/cds/cds-services/my-service"
... )
>>> client.load_from_registry("mimic-on-fhir", sample_size=10)
>>> responses = client.send_requests()

Load CDA file from path:

>>> client = SandboxClient(
...     url="http://localhost:8000/notereader/fhir/",
...     protocol="soap"
... )
>>> client.load_from_path("./data/clinical_note.xml")
>>> responses = client.send_requests()

Generate data from free text:

>>> client = SandboxClient(
...     url="http://localhost:8000/cds/cds-services/discharge-summarizer"
... )
>>> client.load_free_text(
...     csv_path="./data/notes.csv",
...     column_name="text",
...     workflow="encounter-discharge"
... )
>>> responses = client.send_requests()

Source code in healthchain/sandbox/sandboxclient.py

class SandboxClient:
    """
    Simplified client for testing healthcare services with various data sources.

    This class provides an intuitive interface for:
    - Loading test datasets (MIMIC-on-FHIR, Synthea)
    - Generating synthetic FHIR data
    - Sending requests to healthcare services
    - Managing request/response lifecycle

    Examples:
        Load from dataset registry:
        >>> client = SandboxClient(
        ...     url="http://localhost:8000/cds/cds-services/my-service"
        ... )
        >>> client.load_from_registry("mimic-on-fhir", sample_size=10)
        >>> responses = client.send_requests()

        Load CDA file from path:
        >>> client = SandboxClient(
        ...     url="http://localhost:8000/notereader/fhir/",
        ...     protocol="soap"
        ... )
        >>> client.load_from_path("./data/clinical_note.xml")
        >>> responses = client.send_requests()

        Generate data from free text:
        >>> client = SandboxClient(
        ...     url="http://localhost:8000/cds/cds-services/discharge-summarizer"
        ... )
        >>> client.load_free_text(
        ...     csv_path="./data/notes.csv",
        ...     column_name="text",
        ...     workflow="encounter-discharge"
        ... )
        >>> responses = client.send_requests()
    """

    def __init__(
        self,
        url: str,
        workflow: Union[Workflow, str],
        protocol: Literal["rest", "soap"] = "rest",
        timeout: float = 10.0,
    ):
        """
        Initialize SandboxClient.

        Args:
            url: Full service URL (e.g., "http://localhost:8000/cds/cds-services/my-service")
            workflow: Workflow specification (required) - determines request type and validation
            protocol: Communication protocol - "rest" for CDS Hooks, "soap" for CDA
            timeout: Request timeout in seconds

        Raises:
            ValueError: If url or workflow-protocol combination is invalid
        """
        try:
            self.url = httpx.URL(url)
        except Exception as e:
            raise ValueError(f"Invalid URL: {str(e)}")

        self.workflow = Workflow(workflow) if isinstance(workflow, str) else workflow
        self.protocol = ApiProtocol.soap if protocol == "soap" else ApiProtocol.rest
        self.timeout = timeout

        # Request/response management
        self.requests: List[Union[CDSRequest, Any]] = []
        self.responses: List[Dict] = []
        self.sandbox_id = uuid.uuid4()

        # Single validation point - fail fast on incompatible workflow-protocol
        self._validate_workflow_protocol()

        log.info(f"Initialized SandboxClient {self.sandbox_id} for {self.url}")

    def _validate_workflow_protocol(self) -> None:
        """
        Validate workflow is compatible with protocol.

        Raises:
            ValueError: If workflow-protocol combination is invalid
        """
        from healthchain.sandbox.workflows import UseCaseMapping

        if self.protocol == ApiProtocol.soap:
            # SOAP only works with ClinicalDocumentation workflows
            soap_workflows = UseCaseMapping.ClinicalDocumentation.allowed_workflows
            if self.workflow.value not in soap_workflows:
                raise ValueError(
                    f"Workflow '{self.workflow.value}' is not compatible with SOAP protocol. "
                    f"SOAP requires Clinical Documentation workflows: {soap_workflows}"
                )

        elif self.protocol == ApiProtocol.rest:
            # REST only works with CDS workflows
            rest_workflows = UseCaseMapping.ClinicalDecisionSupport.allowed_workflows
            if self.workflow.value not in rest_workflows:
                raise ValueError(
                    f"Workflow '{self.workflow.value}' is not compatible with REST protocol. "
                    f"REST requires CDS workflows: {rest_workflows}"
                )

    def load_from_registry(
        self,
        source: str,
        data_dir: str,
        **kwargs: Any,
    ) -> "SandboxClient":
        """
        Load data from the dataset registry.

        Loads pre-configured datasets like MIMIC-on-FHIR, Synthea, or custom
        registered datasets.

        Args:
            source: Dataset name (e.g., "mimic-on-fhir", "synthea")
            data_dir: Path to the dataset directory
            **kwargs: Dataset-specific parameters (e.g., resource_types, sample_size)

        Returns:
            Self for method chaining

        Raises:
            ValueError: If dataset not found in registry
            FileNotFoundError: If data_dir doesn't exist

        Examples:
            Load MIMIC dataset:
            >>> client = SandboxClient(
            ...     url="http://localhost:8000/cds/patient-view",
            ...     workflow="patient-view",
            ... )
            >>> client.load_from_registry(
            ...     "mimic-on-fhir",
            ...     data_dir="./data/mimic-fhir",
            ...     resource_types=["MimicMedication"],
            ...     sample_size=10
            ... )
        """
        from healthchain.sandbox.datasets import DatasetRegistry

        log.info(f"Loading dataset from registry: {source}")
        try:
            loaded_data = DatasetRegistry.load(source, data_dir=data_dir, **kwargs)
            self._construct_request(loaded_data)
            log.info(f"Loaded {source} dataset with {len(self.requests)} requests")
        except KeyError:
            raise ValueError(
                f"Unknown dataset: {source}. "
                f"Available datasets: {DatasetRegistry.list_datasets()}"
            )
        return self

    def load_from_path(
        self,
        path: Union[str, Path],
        pattern: Optional[str] = None,
    ) -> "SandboxClient":
        """
        Load data from a file or directory.

        Supports single files or all matching files in a directory (with optional glob pattern).
        For .xml (SOAP protocol) loads CDA; for .json (REST protocol) loads Prefetch.

        Args:
            path: File or directory path.
            pattern: Glob pattern for files in directory (e.g., "*.xml").

        Returns:
            Self.

        Raises:
            FileNotFoundError: If path does not exist.
            ValueError: If no matching files are found or if path is not file/dir.
        """

        path = Path(path)
        if not path.exists():
            raise FileNotFoundError(f"Path not found: {path}")

        # Collect files to process
        files_to_load = []
        if path.is_file():
            files_to_load = [path]
        elif path.is_dir():
            pattern = pattern or "*"
            files_to_load = list(path.glob(pattern))
            if not files_to_load:
                raise ValueError(
                    f"No files found matching pattern '{pattern}' in {path}"
                )
        else:
            raise ValueError(f"Path must be a file or directory: {path}")

        log.info(f"Loading {len(files_to_load)} file(s) from {path}")

        # Process each file
        for file_path in files_to_load:
            # Determine file type from extension
            extension = file_path.suffix.lower()

            if extension == ".xml":
                with open(file_path, "r") as f:
                    xml_content = f.read()
                self._construct_request(xml_content)
                log.info(f"Loaded CDA document from {file_path.name}")

            elif extension == ".json":
                with open(file_path, "r") as f:
                    json_data = json.load(f)

                try:
                    self._construct_request(json_data)
                    log.info(f"Loaded prefetch data from {file_path.name}")

                except Exception as e:
                    log.error(f"Failed to parse {file_path} as prefetch data: {e}")
                    raise ValueError(
                        f"File {file_path} is not valid prefetch format. "
                        f"Expected JSON with FHIR resources. "
                        f"Error: {e}"
                    )
            else:
                log.warning(f"Skipping unsupported file type: {file_path}")

        log.info(
            f"Loaded {len(self.requests)} requests from {len(files_to_load)} file(s)"
        )
        return self

    def load_free_text(
        self,
        csv_path: str,
        column_name: str,
        generate_synthetic: bool = True,
        random_seed: Optional[int] = None,
        **kwargs: Any,
    ) -> "SandboxClient":
        """
        Load free-text notes from a CSV column and create FHIR DocumentReferences for CDS prefetch.
        Optionally include synthetic FHIR resources based on the current workflow.

        Args:
            csv_path: Path to the CSV file
            column_name: Name of the text column
            generate_synthetic: Whether to add synthetic FHIR resources (default: True)
            random_seed: Seed for reproducible results
            **kwargs: Extra parameters for data generation

        Returns:
            self

        Raises:
            FileNotFoundError: If the CSV file does not exist
            ValueError: If the column is not found
        """
        from .generators import CdsDataGenerator

        generator = CdsDataGenerator()
        generator.set_workflow(self.workflow)

        prefetch_data = generator.generate_prefetch(
            random_seed=random_seed,
            free_text_path=csv_path,
            column_name=column_name,
            generate_resources=generate_synthetic,
            **kwargs,
        )

        self._construct_request(prefetch_data)

        if generate_synthetic:
            log.info(
                f"Generated {len(self.requests)} requests from free text with synthetic resources for workflow {self.workflow.value}"
            )
        else:
            log.info(
                f"Generated {len(self.requests)} requests from free text only (no synthetic resources)"
            )

        return self

    def _construct_request(self, data: Union[Dict[str, Any], Any]) -> None:
        """
        Convert data to request format and add to queue.

        Args:
            data: Data to convert (Dict for CDS prefetch, string for CDA)
        """
        if self.protocol == ApiProtocol.rest:
            constructor = CdsRequestConstructor()
            request = constructor.construct_request(data, self.workflow)
        elif self.protocol == ApiProtocol.soap:
            constructor = ClinDocRequestConstructor()
            request = constructor.construct_request(data, self.workflow)
        else:
            raise ValueError(f"Unsupported protocol: {self.protocol}")

        self.requests.append(request)

    def clear_requests(self) -> "SandboxClient":
        """
        Clear all queued requests.

        Useful when you want to start fresh without creating a new client instance.

        Returns:
            Self for method chaining
        """
        count = len(self.requests)
        self.requests.clear()
        log.info(f"Cleared {count} queued request(s)")

        return self

    def preview_requests(self, limit: Optional[int] = None) -> List[Dict[str, Any]]:
        """
        Get preview of queued requests without sending them.

        Provides a summary view of what requests are queued, useful for debugging
        and verifying data was loaded correctly before sending.

        Args:
            limit: Maximum number of requests to preview. If None, preview all.

        Returns:
            List of request summary dictionaries containing metadata
        """
        requests = self.requests[:limit] if limit else self.requests
        previews = []

        for idx, req in enumerate(requests):
            preview = {
                "index": idx,
                "type": req.__class__.__name__,
                "protocol": self.protocol.value
                if hasattr(self.protocol, "value")
                else str(self.protocol),
            }

            # Add protocol-specific info
            if self.protocol == ApiProtocol.rest and hasattr(req, "hook"):
                preview["hook"] = req.hook
                preview["hookInstance"] = getattr(req, "hookInstance", None)
            elif self.protocol == ApiProtocol.soap:
                preview["has_document"] = hasattr(req, "document")

            previews.append(preview)

        return previews

    def get_request_data(
        self, format: Literal["dict", "json"] = "dict"
    ) -> Union[List[Dict], str]:
        """
        Get transformed request data for inspection.

        Allows access to serialized request data for debugging or custom processing.
        For direct access to Pydantic models, use the `requests` attribute:
            >>> for request in client.requests:
            >>>     print(request.model_dump())

        Args:
            format: Return format - "dict" for list of dictionaries,
                   "json" for JSON string

        Returns:
            Request data in specified format

        Raises:
            ValueError: If format is not "dict" or "json"

        Examples:
            >>> client.load_from_path("data.xml")
            >>> # Access raw Pydantic models directly
            >>> for request in client.requests:
            >>>     print(request.model_dump(exclude_none=True))
            >>> # Get as dictionaries
            >>> dicts = client.get_request_data("dict")
            >>> # Get as JSON string
            >>> json_str = client.get_request_data("json")
            >>> print(json_str)
        """
        if format == "dict":
            result = []
            for req in self.requests:
                if hasattr(req, "model_dump"):
                    result.append(req.model_dump(exclude_none=True))
                elif hasattr(req, "model_dump_xml"):
                    result.append({"document": req.model_dump_xml()})
                else:
                    result.append(req)
            return result
        elif format == "json":
            return json.dumps(self.get_request_data("dict"), indent=2)
        else:
            raise ValueError(
                f"Invalid format '{format}'. Must be 'dict' or 'json'. "
                f"For raw Pydantic models, access the 'requests' attribute directly."
            )

    def send_requests(self) -> List[Dict]:
        """
        Send all queued requests to the service.

        Returns:
            List of response dictionaries
        """
        if not self.requests:
            raise RuntimeError(
                "No requests to send. Load data first using load_from_registry(), load_from_path(), or load_free_text()"
            )

        log.info(f"Sending {len(self.requests)} requests to {self.url}")

        with httpx.Client(follow_redirects=True) as client:
            responses: List[Dict] = []
            timeout = httpx.Timeout(self.timeout, read=None)

            for request in self.requests:
                try:
                    if self.protocol == ApiProtocol.soap:
                        headers = {"Content-Type": "text/xml; charset=utf-8"}
                        response = client.post(
                            url=str(self.url),
                            data=request.document,
                            headers=headers,
                            timeout=timeout,
                        )
                        response.raise_for_status()
                        response_model = CdaResponse(document=response.text)
                        responses.append(response_model.model_dump_xml())
                    else:
                        # REST/CDS Hooks
                        log.debug(f"Making POST request to: {self.url}")
                        response = client.post(
                            url=str(self.url),
                            json=request.model_dump(exclude_none=True),
                            timeout=timeout,
                        )
                        response.raise_for_status()

                        try:
                            response_data = response.json()
                            cds_response = CDSResponse(**response_data)
                            responses.append(cds_response.model_dump(exclude_none=True))
                        except json.JSONDecodeError:
                            log.error(
                                f"Invalid JSON response from {self.url}. "
                                f"Response preview: {response.text[:200]}"
                            )
                            responses.append({})
                        except Exception:
                            # Fallback to raw response if CDSResponse parsing fails
                            responses.append(response_data)

                except httpx.HTTPStatusError as exc:
                    try:
                        error_content = exc.response.json()
                    except Exception:
                        error_content = exc.response.text
                    log.error(
                        f"Error response {exc.response.status_code} while requesting "
                        f"{exc.request.url!r}: {error_content}"
                    )
                    responses.append({})
                except httpx.TimeoutException as exc:
                    log.error(f"Request to {exc.request.url!r} timed out!")
                    responses.append({})
                except httpx.RequestError as exc:
                    log.error(
                        f"An error occurred while requesting {exc.request.url!r}."
                    )
                    responses.append({})

        self.responses = responses
        log.info(f"Received {len(responses)} responses")

        return responses

    def save_results(self, directory: Union[str, Path] = "./output/") -> None:
        """
        Save request and response data to disk.

        Args:
            directory: Directory to save data to (default: "./output/")
        """
        if not self.responses:
            raise RuntimeError(
                "No responses to save. Send requests first using send_requests()"
            )

        save_dir = Path(directory)
        request_path = ensure_directory_exists(save_dir / "requests")

        # Determine file extension based on protocol
        extension = "xml" if self.protocol == ApiProtocol.soap else "json"

        # Save requests
        if self.protocol == ApiProtocol.soap:
            request_data = [request.model_dump_xml() for request in self.requests]
        else:
            request_data = [
                request.model_dump(exclude_none=True) for request in self.requests
            ]

        save_data_to_directory(
            request_data,
            "request",
            self.sandbox_id,
            request_path,
            extension,
        )
        log.info(f"Saved request data at {request_path}/")

        # Save responses
        response_path = ensure_directory_exists(save_dir / "responses")
        save_data_to_directory(
            self.responses,
            "response",
            self.sandbox_id,
            response_path,
            extension,
        )
        log.info(f"Saved response data at {response_path}/")

    def get_status(self) -> Dict[str, Any]:
        """
        Get current client status and statistics.

        Returns:
            Dictionary containing client status information
        """
        return {
            "sandbox_id": str(self.sandbox_id),
            "url": str(self.url),
            "protocol": self.protocol.value
            if hasattr(self.protocol, "value")
            else str(self.protocol),
            "workflow": self.workflow.value if self.workflow else None,
            "requests_queued": len(self.requests),
            "responses_received": len(self.responses),
        }

    def __enter__(self) -> "SandboxClient":
        """Context manager entry."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        """
        Context manager exit - auto-save results if responses exist.

        Only saves if no exception occurred and responses were generated.
        """
        if self.responses and exc_type is None:
            try:
                self.save_results()
                log.info("Auto-saved results on context exit")
            except Exception as e:
                log.warning(f"Failed to auto-save results: {e}")

    def __repr__(self) -> str:
        """String representation of SandboxClient."""
        return (
            f"SandboxClient(url='{self.url}', "
            f"protocol='{self.protocol.value if hasattr(self.protocol, 'value') else self.protocol}', "
            f"requests={len(self.requests)})"
        )

__enter__()

Context manager entry.

Source code in healthchain/sandbox/sandboxclient.py

def __enter__(self) -> "SandboxClient":
    """Context manager entry."""
    return self

__exit__(exc_type, exc_val, exc_tb)

Context manager exit - auto-save results if responses exist.

Only saves if no exception occurred and responses were generated.

Source code in healthchain/sandbox/sandboxclient.py

def __exit__(self, exc_type, exc_val, exc_tb) -> None:
    """
    Context manager exit - auto-save results if responses exist.

    Only saves if no exception occurred and responses were generated.
    """
    if self.responses and exc_type is None:
        try:
            self.save_results()
            log.info("Auto-saved results on context exit")
        except Exception as e:
            log.warning(f"Failed to auto-save results: {e}")

__init__(url, workflow, protocol='rest', timeout=10.0)

Initialize SandboxClient.

PARAMETER	DESCRIPTION
url	Full service URL (e.g., "http://localhost:8000/cds/cds-services/my-service") TYPE: str
workflow	Workflow specification (required) - determines request type and validation TYPE: Union[Workflow, str]
protocol	Communication protocol - "rest" for CDS Hooks, "soap" for CDA TYPE: Literal['rest', 'soap'] DEFAULT: 'rest'
timeout	Request timeout in seconds TYPE: float DEFAULT: 10.0

RAISES	DESCRIPTION
ValueError	If url or workflow-protocol combination is invalid

Source code in healthchain/sandbox/sandboxclient.py

def __init__(
    self,
    url: str,
    workflow: Union[Workflow, str],
    protocol: Literal["rest", "soap"] = "rest",
    timeout: float = 10.0,
):
    """
    Initialize SandboxClient.

    Args:
        url: Full service URL (e.g., "http://localhost:8000/cds/cds-services/my-service")
        workflow: Workflow specification (required) - determines request type and validation
        protocol: Communication protocol - "rest" for CDS Hooks, "soap" for CDA
        timeout: Request timeout in seconds

    Raises:
        ValueError: If url or workflow-protocol combination is invalid
    """
    try:
        self.url = httpx.URL(url)
    except Exception as e:
        raise ValueError(f"Invalid URL: {str(e)}")

    self.workflow = Workflow(workflow) if isinstance(workflow, str) else workflow
    self.protocol = ApiProtocol.soap if protocol == "soap" else ApiProtocol.rest
    self.timeout = timeout

    # Request/response management
    self.requests: List[Union[CDSRequest, Any]] = []
    self.responses: List[Dict] = []
    self.sandbox_id = uuid.uuid4()

    # Single validation point - fail fast on incompatible workflow-protocol
    self._validate_workflow_protocol()

    log.info(f"Initialized SandboxClient {self.sandbox_id} for {self.url}")

__repr__()

String representation of SandboxClient.

Source code in healthchain/sandbox/sandboxclient.py

def __repr__(self) -> str:
    """String representation of SandboxClient."""
    return (
        f"SandboxClient(url='{self.url}', "
        f"protocol='{self.protocol.value if hasattr(self.protocol, 'value') else self.protocol}', "
        f"requests={len(self.requests)})"
    )

clear_requests()

Clear all queued requests.

Useful when you want to start fresh without creating a new client instance.

RETURNS	DESCRIPTION
SandboxClient	Self for method chaining

Source code in healthchain/sandbox/sandboxclient.py

def clear_requests(self) -> "SandboxClient":
    """
    Clear all queued requests.

    Useful when you want to start fresh without creating a new client instance.

    Returns:
        Self for method chaining
    """
    count = len(self.requests)
    self.requests.clear()
    log.info(f"Cleared {count} queued request(s)")

    return self

get_request_data(format='dict')

Get transformed request data for inspection.

Allows access to serialized request data for debugging or custom processing. For direct access to Pydantic models, use the requests attribute: >>> for request in client.requests: >>> print(request.model_dump())

PARAMETER	DESCRIPTION
format	Return format - "dict" for list of dictionaries, "json" for JSON string TYPE: Literal['dict', 'json'] DEFAULT: 'dict'

RETURNS	DESCRIPTION
Union[List[Dict], str]	Request data in specified format

RAISES	DESCRIPTION
ValueError	If format is not "dict" or "json"

Examples:

>>> client.load_from_path("data.xml")
>>> # Access raw Pydantic models directly
>>> for request in client.requests:
>>>     print(request.model_dump(exclude_none=True))
>>> # Get as dictionaries
>>> dicts = client.get_request_data("dict")
>>> # Get as JSON string
>>> json_str = client.get_request_data("json")
>>> print(json_str)

Source code in healthchain/sandbox/sandboxclient.py

def get_request_data(
    self, format: Literal["dict", "json"] = "dict"
) -> Union[List[Dict], str]:
    """
    Get transformed request data for inspection.

    Allows access to serialized request data for debugging or custom processing.
    For direct access to Pydantic models, use the `requests` attribute:
        >>> for request in client.requests:
        >>>     print(request.model_dump())

    Args:
        format: Return format - "dict" for list of dictionaries,
               "json" for JSON string

    Returns:
        Request data in specified format

    Raises:
        ValueError: If format is not "dict" or "json"

    Examples:
        >>> client.load_from_path("data.xml")
        >>> # Access raw Pydantic models directly
        >>> for request in client.requests:
        >>>     print(request.model_dump(exclude_none=True))
        >>> # Get as dictionaries
        >>> dicts = client.get_request_data("dict")
        >>> # Get as JSON string
        >>> json_str = client.get_request_data("json")
        >>> print(json_str)
    """
    if format == "dict":
        result = []
        for req in self.requests:
            if hasattr(req, "model_dump"):
                result.append(req.model_dump(exclude_none=True))
            elif hasattr(req, "model_dump_xml"):
                result.append({"document": req.model_dump_xml()})
            else:
                result.append(req)
        return result
    elif format == "json":
        return json.dumps(self.get_request_data("dict"), indent=2)
    else:
        raise ValueError(
            f"Invalid format '{format}'. Must be 'dict' or 'json'. "
            f"For raw Pydantic models, access the 'requests' attribute directly."
        )

get_status()

Get current client status and statistics.

RETURNS	DESCRIPTION
Dict[str, Any]	Dictionary containing client status information

Source code in healthchain/sandbox/sandboxclient.py

def get_status(self) -> Dict[str, Any]:
    """
    Get current client status and statistics.

    Returns:
        Dictionary containing client status information
    """
    return {
        "sandbox_id": str(self.sandbox_id),
        "url": str(self.url),
        "protocol": self.protocol.value
        if hasattr(self.protocol, "value")
        else str(self.protocol),
        "workflow": self.workflow.value if self.workflow else None,
        "requests_queued": len(self.requests),
        "responses_received": len(self.responses),
    }

load_free_text(csv_path, column_name, generate_synthetic=True, random_seed=None, **kwargs)

Load free-text notes from a CSV column and create FHIR DocumentReferences for CDS prefetch. Optionally include synthetic FHIR resources based on the current workflow.

PARAMETER	DESCRIPTION
csv_path	Path to the CSV file TYPE: str
column_name	Name of the text column TYPE: str
generate_synthetic	Whether to add synthetic FHIR resources (default: True) TYPE: bool DEFAULT: True
random_seed	Seed for reproducible results TYPE: Optional[int] DEFAULT: None
**kwargs	Extra parameters for data generation TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
SandboxClient	self

RAISES	DESCRIPTION
FileNotFoundError	If the CSV file does not exist
ValueError	If the column is not found

Source code in healthchain/sandbox/sandboxclient.py

def load_free_text(
    self,
    csv_path: str,
    column_name: str,
    generate_synthetic: bool = True,
    random_seed: Optional[int] = None,
    **kwargs: Any,
) -> "SandboxClient":
    """
    Load free-text notes from a CSV column and create FHIR DocumentReferences for CDS prefetch.
    Optionally include synthetic FHIR resources based on the current workflow.

    Args:
        csv_path: Path to the CSV file
        column_name: Name of the text column
        generate_synthetic: Whether to add synthetic FHIR resources (default: True)
        random_seed: Seed for reproducible results
        **kwargs: Extra parameters for data generation

    Returns:
        self

    Raises:
        FileNotFoundError: If the CSV file does not exist
        ValueError: If the column is not found
    """
    from .generators import CdsDataGenerator

    generator = CdsDataGenerator()
    generator.set_workflow(self.workflow)

    prefetch_data = generator.generate_prefetch(
        random_seed=random_seed,
        free_text_path=csv_path,
        column_name=column_name,
        generate_resources=generate_synthetic,
        **kwargs,
    )

    self._construct_request(prefetch_data)

    if generate_synthetic:
        log.info(
            f"Generated {len(self.requests)} requests from free text with synthetic resources for workflow {self.workflow.value}"
        )
    else:
        log.info(
            f"Generated {len(self.requests)} requests from free text only (no synthetic resources)"
        )

    return self

load_from_path(path, pattern=None)

Load data from a file or directory.

Supports single files or all matching files in a directory (with optional glob pattern). For .xml (SOAP protocol) loads CDA; for .json (REST protocol) loads Prefetch.

PARAMETER	DESCRIPTION
path	File or directory path. TYPE: Union[str, Path]
pattern	Glob pattern for files in directory (e.g., "*.xml"). TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
SandboxClient	Self.

RAISES	DESCRIPTION
FileNotFoundError	If path does not exist.
ValueError	If no matching files are found or if path is not file/dir.

Source code in healthchain/sandbox/sandboxclient.py

def load_from_path(
    self,
    path: Union[str, Path],
    pattern: Optional[str] = None,
) -> "SandboxClient":
    """
    Load data from a file or directory.

    Supports single files or all matching files in a directory (with optional glob pattern).
    For .xml (SOAP protocol) loads CDA; for .json (REST protocol) loads Prefetch.

    Args:
        path: File or directory path.
        pattern: Glob pattern for files in directory (e.g., "*.xml").

    Returns:
        Self.

    Raises:
        FileNotFoundError: If path does not exist.
        ValueError: If no matching files are found or if path is not file/dir.
    """

    path = Path(path)
    if not path.exists():
        raise FileNotFoundError(f"Path not found: {path}")

    # Collect files to process
    files_to_load = []
    if path.is_file():
        files_to_load = [path]
    elif path.is_dir():
        pattern = pattern or "*"
        files_to_load = list(path.glob(pattern))
        if not files_to_load:
            raise ValueError(
                f"No files found matching pattern '{pattern}' in {path}"
            )
    else:
        raise ValueError(f"Path must be a file or directory: {path}")

    log.info(f"Loading {len(files_to_load)} file(s) from {path}")

    # Process each file
    for file_path in files_to_load:
        # Determine file type from extension
        extension = file_path.suffix.lower()

        if extension == ".xml":
            with open(file_path, "r") as f:
                xml_content = f.read()
            self._construct_request(xml_content)
            log.info(f"Loaded CDA document from {file_path.name}")

        elif extension == ".json":
            with open(file_path, "r") as f:
                json_data = json.load(f)

            try:
                self._construct_request(json_data)
                log.info(f"Loaded prefetch data from {file_path.name}")

            except Exception as e:
                log.error(f"Failed to parse {file_path} as prefetch data: {e}")
                raise ValueError(
                    f"File {file_path} is not valid prefetch format. "
                    f"Expected JSON with FHIR resources. "
                    f"Error: {e}"
                )
        else:
            log.warning(f"Skipping unsupported file type: {file_path}")

    log.info(
        f"Loaded {len(self.requests)} requests from {len(files_to_load)} file(s)"
    )
    return self

load_from_registry(source, data_dir, **kwargs)

Load data from the dataset registry.

Loads pre-configured datasets like MIMIC-on-FHIR, Synthea, or custom registered datasets.

PARAMETER	DESCRIPTION
source	Dataset name (e.g., "mimic-on-fhir", "synthea") TYPE: str
data_dir	Path to the dataset directory TYPE: str
**kwargs	Dataset-specific parameters (e.g., resource_types, sample_size) TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
SandboxClient	Self for method chaining

RAISES	DESCRIPTION
ValueError	If dataset not found in registry
FileNotFoundError	If data_dir doesn't exist

Examples:

Load MIMIC dataset:

>>> client = SandboxClient(
...     url="http://localhost:8000/cds/patient-view",
...     workflow="patient-view",
... )
>>> client.load_from_registry(
...     "mimic-on-fhir",
...     data_dir="./data/mimic-fhir",
...     resource_types=["MimicMedication"],
...     sample_size=10
... )

Source code in healthchain/sandbox/sandboxclient.py

def load_from_registry(
    self,
    source: str,
    data_dir: str,
    **kwargs: Any,
) -> "SandboxClient":
    """
    Load data from the dataset registry.

    Loads pre-configured datasets like MIMIC-on-FHIR, Synthea, or custom
    registered datasets.

    Args:
        source: Dataset name (e.g., "mimic-on-fhir", "synthea")
        data_dir: Path to the dataset directory
        **kwargs: Dataset-specific parameters (e.g., resource_types, sample_size)

    Returns:
        Self for method chaining

    Raises:
        ValueError: If dataset not found in registry
        FileNotFoundError: If data_dir doesn't exist

    Examples:
        Load MIMIC dataset:
        >>> client = SandboxClient(
        ...     url="http://localhost:8000/cds/patient-view",
        ...     workflow="patient-view",
        ... )
        >>> client.load_from_registry(
        ...     "mimic-on-fhir",
        ...     data_dir="./data/mimic-fhir",
        ...     resource_types=["MimicMedication"],
        ...     sample_size=10
        ... )
    """
    from healthchain.sandbox.datasets import DatasetRegistry

    log.info(f"Loading dataset from registry: {source}")
    try:
        loaded_data = DatasetRegistry.load(source, data_dir=data_dir, **kwargs)
        self._construct_request(loaded_data)
        log.info(f"Loaded {source} dataset with {len(self.requests)} requests")
    except KeyError:
        raise ValueError(
            f"Unknown dataset: {source}. "
            f"Available datasets: {DatasetRegistry.list_datasets()}"
        )
    return self

preview_requests(limit=None)

Get preview of queued requests without sending them.

Provides a summary view of what requests are queued, useful for debugging and verifying data was loaded correctly before sending.

PARAMETER	DESCRIPTION
limit	Maximum number of requests to preview. If None, preview all. TYPE: Optional[int] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict[str, Any]]	List of request summary dictionaries containing metadata

Source code in healthchain/sandbox/sandboxclient.py

def preview_requests(self, limit: Optional[int] = None) -> List[Dict[str, Any]]:
    """
    Get preview of queued requests without sending them.

    Provides a summary view of what requests are queued, useful for debugging
    and verifying data was loaded correctly before sending.

    Args:
        limit: Maximum number of requests to preview. If None, preview all.

    Returns:
        List of request summary dictionaries containing metadata
    """
    requests = self.requests[:limit] if limit else self.requests
    previews = []

    for idx, req in enumerate(requests):
        preview = {
            "index": idx,
            "type": req.__class__.__name__,
            "protocol": self.protocol.value
            if hasattr(self.protocol, "value")
            else str(self.protocol),
        }

        # Add protocol-specific info
        if self.protocol == ApiProtocol.rest and hasattr(req, "hook"):
            preview["hook"] = req.hook
            preview["hookInstance"] = getattr(req, "hookInstance", None)
        elif self.protocol == ApiProtocol.soap:
            preview["has_document"] = hasattr(req, "document")

        previews.append(preview)

    return previews

save_results(directory='./output/')

Save request and response data to disk.

PARAMETER	DESCRIPTION
directory	Directory to save data to (default: "./output/") TYPE: Union[str, Path] DEFAULT: './output/'

Source code in healthchain/sandbox/sandboxclient.py

def save_results(self, directory: Union[str, Path] = "./output/") -> None:
    """
    Save request and response data to disk.

    Args:
        directory: Directory to save data to (default: "./output/")
    """
    if not self.responses:
        raise RuntimeError(
            "No responses to save. Send requests first using send_requests()"
        )

    save_dir = Path(directory)
    request_path = ensure_directory_exists(save_dir / "requests")

    # Determine file extension based on protocol
    extension = "xml" if self.protocol == ApiProtocol.soap else "json"

    # Save requests
    if self.protocol == ApiProtocol.soap:
        request_data = [request.model_dump_xml() for request in self.requests]
    else:
        request_data = [
            request.model_dump(exclude_none=True) for request in self.requests
        ]

    save_data_to_directory(
        request_data,
        "request",
        self.sandbox_id,
        request_path,
        extension,
    )
    log.info(f"Saved request data at {request_path}/")

    # Save responses
    response_path = ensure_directory_exists(save_dir / "responses")
    save_data_to_directory(
        self.responses,
        "response",
        self.sandbox_id,
        response_path,
        extension,
    )
    log.info(f"Saved response data at {response_path}/")

send_requests()

Send all queued requests to the service.

RETURNS	DESCRIPTION
List[Dict]	List of response dictionaries

Source code in healthchain/sandbox/sandboxclient.py

def send_requests(self) -> List[Dict]:
    """
    Send all queued requests to the service.

    Returns:
        List of response dictionaries
    """
    if not self.requests:
        raise RuntimeError(
            "No requests to send. Load data first using load_from_registry(), load_from_path(), or load_free_text()"
        )

    log.info(f"Sending {len(self.requests)} requests to {self.url}")

    with httpx.Client(follow_redirects=True) as client:
        responses: List[Dict] = []
        timeout = httpx.Timeout(self.timeout, read=None)

        for request in self.requests:
            try:
                if self.protocol == ApiProtocol.soap:
                    headers = {"Content-Type": "text/xml; charset=utf-8"}
                    response = client.post(
                        url=str(self.url),
                        data=request.document,
                        headers=headers,
                        timeout=timeout,
                    )
                    response.raise_for_status()
                    response_model = CdaResponse(document=response.text)
                    responses.append(response_model.model_dump_xml())
                else:
                    # REST/CDS Hooks
                    log.debug(f"Making POST request to: {self.url}")
                    response = client.post(
                        url=str(self.url),
                        json=request.model_dump(exclude_none=True),
                        timeout=timeout,
                    )
                    response.raise_for_status()

                    try:
                        response_data = response.json()
                        cds_response = CDSResponse(**response_data)
                        responses.append(cds_response.model_dump(exclude_none=True))
                    except json.JSONDecodeError:
                        log.error(
                            f"Invalid JSON response from {self.url}. "
                            f"Response preview: {response.text[:200]}"
                        )
                        responses.append({})
                    except Exception:
                        # Fallback to raw response if CDSResponse parsing fails
                        responses.append(response_data)

            except httpx.HTTPStatusError as exc:
                try:
                    error_content = exc.response.json()
                except Exception:
                    error_content = exc.response.text
                log.error(
                    f"Error response {exc.response.status_code} while requesting "
                    f"{exc.request.url!r}: {error_content}"
                )
                responses.append({})
            except httpx.TimeoutException as exc:
                log.error(f"Request to {exc.request.url!r} timed out!")
                responses.append({})
            except httpx.RequestError as exc:
                log.error(
                    f"An error occurred while requesting {exc.request.url!r}."
                )
                responses.append({})

    self.responses = responses
    log.info(f"Received {len(responses)} responses")

    return responses

CdsDataGenerator

Generates synthetic CDS (Clinical Decision Support) data for specified workflows.

Uses registered generators to create FHIR resources (e.g., Patient, Encounter, Condition) according to workflow configuration. Can optionally include free text data from a CSV file as DocumentReference.

ATTRIBUTE	DESCRIPTION
registry	Maps generator names to classes. TYPE: dict
mappings	Maps workflows to required generators. TYPE: dict
generated_data	Most recently generated resources. TYPE: Dict[str, Resource]
workflow	Currently active workflow. TYPE: str

Example

generator = CdsDataGenerator() generator.set_workflow("encounter_discharge") data = generator.generate_prefetch(random_seed=42)

Source code in healthchain/sandbox/generators/cdsdatagenerator.py

class CdsDataGenerator:
    """
    Generates synthetic CDS (Clinical Decision Support) data for specified workflows.

    Uses registered generators to create FHIR resources (e.g., Patient, Encounter, Condition) according to workflow configuration.
    Can optionally include free text data from a CSV file as DocumentReference.

    Attributes:
        registry (dict): Maps generator names to classes.
        mappings (dict): Maps workflows to required generators.
        generated_data (Dict[str, Resource]): Most recently generated resources.
        workflow (str): Currently active workflow.

    Example:
        >>> generator = CdsDataGenerator()
        >>> generator.set_workflow("encounter_discharge")
        >>> data = generator.generate_prefetch(random_seed=42)
    """

    # TODO: Add ordering and logic so that patient/encounter IDs are passed to subsequent generators
    # TODO: Some of the resources should be allowed to be multiplied

    default_workflow_mappings = {
        Workflow.encounter_discharge: [
            {"generator": "EncounterGenerator"},
            {"generator": "ConditionGenerator"},
            {"generator": "ProcedureGenerator"},
            {"generator": "MedicationRequestGenerator"},
        ],
        Workflow.patient_view: [
            {"generator": "PatientGenerator"},
            {"generator": "EncounterGenerator"},
            {"generator": "ConditionGenerator"},
        ],
    }

    def __init__(self):
        self.registry = generator_registry
        self.mappings = self.default_workflow_mappings
        self.generated_data: Dict[str, Resource] = {}

    def fetch_generator(self, generator_name: str) -> Callable:
        """
        Return the generator class by name from the registry.

        Args:
            generator_name (str): Name of the data generator.

        Returns:
            Callable: Generator class, or None if not found.

        Example:
            >>> gen = CdsDataGenerator().fetch_generator("PatientGenerator")
        """
        return self.registry.get(generator_name)

    def set_workflow(self, workflow: str) -> None:
        """
        Set the current workflow name to use for data generation.

        Args:
            workflow (str): Workflow name.
        """
        self.workflow = workflow

    def generate_prefetch(
        self,
        constraints: Optional[list] = None,
        free_text_path: Optional[str] = None,
        column_name: Optional[str] = None,
        random_seed: Optional[int] = None,
        generate_resources: bool = True,
    ) -> Dict[str, Resource]:
        """
        Generate prefetch FHIR resources and/or DocumentReference.

        Args:
            constraints (Optional[list]): Constraints for resource generation.
            free_text_path (Optional[str]): CSV file containing free text.
            column_name (Optional[str]): CSV column for free text.
            random_seed (Optional[int]): Random seed.
            generate_resources (bool): If True, generate synthetic FHIR resources.

        Returns:
            Dict[str, Resource]: Generated resources keyed by resource type (lowercase), plus "document" if a free text entry is used.

        Raises:
            ValueError: If workflow is not recognized, or column name is missing.
            FileNotFoundError: If free_text_path does not exist.
        """
        prefetch = {}

        if generate_resources:
            if self.workflow not in self.mappings:
                raise ValueError(f"Workflow {self.workflow} not found in mappings")

            for resource in self.mappings[self.workflow]:
                generator_name = resource["generator"]
                generator = self.fetch_generator(generator_name)
                resource = generator.generate(
                    constraints=constraints, random_seed=random_seed
                )
                prefetch[resource.__resource_type__.lower()] = resource

        parsed_free_text = (
            self.free_text_parser(free_text_path, column_name)
            if free_text_path
            else None
        )
        if parsed_free_text:
            prefetch["document"] = create_document_reference(
                data=random.choice(parsed_free_text),
                content_type="text/plain",
                status="current",
                description="Free text created by HealthChain CdsDataGenerator",
                attachment_title="Free text created by HealthChain CdsDataGenerator",
            )

        self.generated_data = prefetch

        return self.generated_data

    def free_text_parser(self, path_to_csv: str, column_name: str) -> List[str]:
        """
        Read a column of free text from a CSV file.

        Args:
            path_to_csv (str): Path to CSV file.
            column_name (str): Column name to extract.

        Returns:
            List[str]: Extracted text values.

        Raises:
            FileNotFoundError: If CSV file does not exist.
            ValueError: If column_name is not provided.
        """
        text_data = []

        path = Path(path_to_csv)
        if not path.is_file():
            raise FileNotFoundError(
                f"The file {path_to_csv} does not exist or is not a file."
            )

        try:
            with path.open(mode="r", newline="") as file:
                reader = csv.DictReader(file)
                if column_name is not None:
                    for row in reader:
                        text_data.append(row[column_name])
                else:
                    raise ValueError(
                        "Column name must be provided when header is True."
                    )
        except Exception as ex:
            logger.error(f"An error occurred: {ex}")

        return text_data

fetch_generator(generator_name)

Return the generator class by name from the registry.

PARAMETER	DESCRIPTION
generator_name	Name of the data generator. TYPE: str

RETURNS	DESCRIPTION
Callable	Generator class, or None if not found. TYPE: Callable

Example

gen = CdsDataGenerator().fetch_generator("PatientGenerator")

Source code in healthchain/sandbox/generators/cdsdatagenerator.py

def fetch_generator(self, generator_name: str) -> Callable:
    """
    Return the generator class by name from the registry.

    Args:
        generator_name (str): Name of the data generator.

    Returns:
        Callable: Generator class, or None if not found.

    Example:
        >>> gen = CdsDataGenerator().fetch_generator("PatientGenerator")
    """
    return self.registry.get(generator_name)

free_text_parser(path_to_csv, column_name)

Read a column of free text from a CSV file.

PARAMETER	DESCRIPTION
path_to_csv	Path to CSV file. TYPE: str
column_name	Column name to extract. TYPE: str

RETURNS	DESCRIPTION
List[str]	List[str]: Extracted text values.

RAISES	DESCRIPTION
FileNotFoundError	If CSV file does not exist.
ValueError	If column_name is not provided.

Source code in healthchain/sandbox/generators/cdsdatagenerator.py

def free_text_parser(self, path_to_csv: str, column_name: str) -> List[str]:
    """
    Read a column of free text from a CSV file.

    Args:
        path_to_csv (str): Path to CSV file.
        column_name (str): Column name to extract.

    Returns:
        List[str]: Extracted text values.

    Raises:
        FileNotFoundError: If CSV file does not exist.
        ValueError: If column_name is not provided.
    """
    text_data = []

    path = Path(path_to_csv)
    if not path.is_file():
        raise FileNotFoundError(
            f"The file {path_to_csv} does not exist or is not a file."
        )

    try:
        with path.open(mode="r", newline="") as file:
            reader = csv.DictReader(file)
            if column_name is not None:
                for row in reader:
                    text_data.append(row[column_name])
            else:
                raise ValueError(
                    "Column name must be provided when header is True."
                )
    except Exception as ex:
        logger.error(f"An error occurred: {ex}")

    return text_data

generate_prefetch(constraints=None, free_text_path=None, column_name=None, random_seed=None, generate_resources=True)

Generate prefetch FHIR resources and/or DocumentReference.

PARAMETER	DESCRIPTION
constraints	Constraints for resource generation. TYPE: Optional[list] DEFAULT: None
free_text_path	CSV file containing free text. TYPE: Optional[str] DEFAULT: None
column_name	CSV column for free text. TYPE: Optional[str] DEFAULT: None
random_seed	Random seed. TYPE: Optional[int] DEFAULT: None
generate_resources	If True, generate synthetic FHIR resources. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
Dict[str, Resource]	Dict[str, Resource]: Generated resources keyed by resource type (lowercase), plus "document" if a free text entry is used.

RAISES	DESCRIPTION
ValueError	If workflow is not recognized, or column name is missing.
FileNotFoundError	If free_text_path does not exist.

Source code in healthchain/sandbox/generators/cdsdatagenerator.py

def generate_prefetch(
    self,
    constraints: Optional[list] = None,
    free_text_path: Optional[str] = None,
    column_name: Optional[str] = None,
    random_seed: Optional[int] = None,
    generate_resources: bool = True,
) -> Dict[str, Resource]:
    """
    Generate prefetch FHIR resources and/or DocumentReference.

    Args:
        constraints (Optional[list]): Constraints for resource generation.
        free_text_path (Optional[str]): CSV file containing free text.
        column_name (Optional[str]): CSV column for free text.
        random_seed (Optional[int]): Random seed.
        generate_resources (bool): If True, generate synthetic FHIR resources.

    Returns:
        Dict[str, Resource]: Generated resources keyed by resource type (lowercase), plus "document" if a free text entry is used.

    Raises:
        ValueError: If workflow is not recognized, or column name is missing.
        FileNotFoundError: If free_text_path does not exist.
    """
    prefetch = {}

    if generate_resources:
        if self.workflow not in self.mappings:
            raise ValueError(f"Workflow {self.workflow} not found in mappings")

        for resource in self.mappings[self.workflow]:
            generator_name = resource["generator"]
            generator = self.fetch_generator(generator_name)
            resource = generator.generate(
                constraints=constraints, random_seed=random_seed
            )
            prefetch[resource.__resource_type__.lower()] = resource

    parsed_free_text = (
        self.free_text_parser(free_text_path, column_name)
        if free_text_path
        else None
    )
    if parsed_free_text:
        prefetch["document"] = create_document_reference(
            data=random.choice(parsed_free_text),
            content_type="text/plain",
            status="current",
            description="Free text created by HealthChain CdsDataGenerator",
            attachment_title="Free text created by HealthChain CdsDataGenerator",
        )

    self.generated_data = prefetch

    return self.generated_data

set_workflow(workflow)

Set the current workflow name to use for data generation.

PARAMETER	DESCRIPTION
workflow	Workflow name. TYPE: str

Source code in healthchain/sandbox/generators/cdsdatagenerator.py

def set_workflow(self, workflow: str) -> None:
    """
    Set the current workflow name to use for data generation.

    Args:
        workflow (str): Workflow name.
    """
    self.workflow = workflow

CDSRequest

Bases: BaseModel

A model representing the data structure for a CDS service call, triggered by specific hooks within a healthcare application.

ATTRIBUTE	DESCRIPTION
hook	The hook that triggered this CDS Service call. For example, 'patient-view'. TYPE: str
hookInstance	A universally unique identifier for this particular hook call. TYPE: UUID
fhirServer	The base URL of the CDS Client's FHIR server. This field is required if fhirAuthorization is provided. TYPE: HttpUrl
fhirAuthorization	Optional authorization details providing a bearer access token for FHIR resources. TYPE: Optional[FhirAuthorization]
context	Hook-specific contextual data required by the CDS service. TYPE: Dict[str, Any]
prefetch	Optional FHIR data that was prefetched by the CDS Client. TYPE: Optional[Dict[str, Any]]

Documentation: https://cds-hooks.org/specification/current/#http-request_1

Source code in healthchain/models/requests/cdsrequest.py

class CDSRequest(BaseModel):
    """
    A model representing the data structure for a CDS service call, triggered by specific hooks
    within a healthcare application.

    Attributes:
        hook (str): The hook that triggered this CDS Service call. For example, 'patient-view'.
        hookInstance (UUID): A universally unique identifier for this particular hook call.
        fhirServer (HttpUrl): The base URL of the CDS Client's FHIR server. This field is required if `fhirAuthorization` is provided.
        fhirAuthorization (Optional[FhirAuthorization]): Optional authorization details providing a bearer access token for FHIR resources.
        context (Dict[str, Any]): Hook-specific contextual data required by the CDS service.
        prefetch (Optional[Dict[str, Any]]): Optional FHIR data that was prefetched by the CDS Client.

    Documentation: https://cds-hooks.org/specification/current/#http-request_1
    """

    hook: str
    hookInstance: str = Field(default_factory=id_generator.generate_random_uuid)
    context: BaseHookContext
    fhirServer: Optional[HttpUrl] = None
    fhirAuthorization: Optional[FHIRAuthorization] = (
        None  # TODO: note this is required if fhirserver is given
    )
    prefetch: Optional[Dict[str, Any]] = None
    extension: Optional[List[Dict[str, Any]]] = None

    def model_dump(self, **kwargs):
        """
        Model dump method to convert any nested datetime and byte objects to strings for readability.
        This is also a workaround to this Pydantic V2 issue https://github.com/pydantic/pydantic/issues/9571
        For proper JSON serialization, should use model_dump_json() instead when issue is resolved.
        """

        def convert_objects(obj):
            if isinstance(obj, dict):
                return {k: convert_objects(v) for k, v in obj.items()}
            elif isinstance(obj, list):
                return [convert_objects(i) for i in obj]
            elif isinstance(obj, datetime):
                return obj.astimezone().isoformat()
            elif isinstance(obj, bytes):
                return obj.decode("utf-8")
            return obj

        dump = super().model_dump(**kwargs)
        return convert_objects(dump)

model_dump(**kwargs)

Model dump method to convert any nested datetime and byte objects to strings for readability. This is also a workaround to this Pydantic V2 issue https://github.com/pydantic/pydantic/issues/9571 For proper JSON serialization, should use model_dump_json() instead when issue is resolved.

Source code in healthchain/models/requests/cdsrequest.py

def model_dump(self, **kwargs):
    """
    Model dump method to convert any nested datetime and byte objects to strings for readability.
    This is also a workaround to this Pydantic V2 issue https://github.com/pydantic/pydantic/issues/9571
    For proper JSON serialization, should use model_dump_json() instead when issue is resolved.
    """

    def convert_objects(obj):
        if isinstance(obj, dict):
            return {k: convert_objects(v) for k, v in obj.items()}
        elif isinstance(obj, list):
            return [convert_objects(i) for i in obj]
        elif isinstance(obj, datetime):
            return obj.astimezone().isoformat()
        elif isinstance(obj, bytes):
            return obj.decode("utf-8")
        return obj

    dump = super().model_dump(**kwargs)
    return convert_objects(dump)

Action

Bases: BaseModel

Within a suggestion, all actions are logically AND'd together, such that a user selecting a suggestion selects all of the actions within it. When a suggestion contains multiple actions, the actions SHOULD be processed as per FHIR's rules for processing transactions with the CDS Client's fhirServer as the base url for the inferred full URL of the transaction bundle entries.

https://cds-hooks.org/specification/current/#action

Source code in healthchain/models/responses/cdsresponse.py

class Action(BaseModel):
    """
    Within a suggestion, all actions are logically AND'd together, such that a user selecting a
    suggestion selects all of the actions within it. When a suggestion contains multiple actions,
    the actions SHOULD be processed as per FHIR's rules for processing transactions with the CDS
    Client's fhirServer as the base url for the inferred full URL of the transaction bundle entries.

    https://cds-hooks.org/specification/current/#action
    """

    type: ActionTypeEnum
    description: str
    resource: Optional[Dict] = None
    resourceId: Optional[str] = None

    @model_validator(mode="after")
    def validate_action_type(self) -> Self:
        if self.type in [ActionTypeEnum.create, ActionTypeEnum.update]:
            assert (
                self.resource
            ), f"'resource' must be provided when type is '{self.type.value}'"
        else:
            assert (
                self.resourceId
            ), f"'resourceId' must be provided when type is '{self.type.value}'"

        return self

ActionTypeEnum

Bases: str, Enum

The type of action being performed

Source code in healthchain/models/responses/cdsresponse.py

class ActionTypeEnum(str, Enum):
    """
    The type of action being performed
    """

    create = "create"
    update = "update"
    delete = "delete"

CDSResponse

Bases: BaseModel

Represents the response from a CDS service.

This class models the structure of a CDS Hooks response, which includes cards for displaying information or suggestions to the user, and optional system actions that can be executed automatically.

ATTRIBUTE	DESCRIPTION
cards	A list of Card objects to be displayed to the end user. Default is an empty list. TYPE: List[Card]
systemActions	A list of Action objects representing actions that the CDS Client should execute as part of performing the decision support requested. This field is optional. TYPE: Optional[List[Action]]

For more information, see: https://cds-hooks.org/specification/current/#cds-service-response

Source code in healthchain/models/responses/cdsresponse.py

class CDSResponse(BaseModel):
    """
    Represents the response from a CDS service.

    This class models the structure of a CDS Hooks response, which includes
    cards for displaying information or suggestions to the user, and optional
    system actions that can be executed automatically.

    Attributes:
        cards (List[Card]): A list of Card objects to be displayed to the end user.
            Default is an empty list.
        systemActions (Optional[List[Action]]): A list of Action objects representing
            actions that the CDS Client should execute as part of performing
            the decision support requested. This field is optional.

    For more information, see:
    https://cds-hooks.org/specification/current/#cds-service-response
    """

    cards: List[Card] = []
    systemActions: Optional[List[Action]] = None

Card

Bases: BaseModel

Cards can provide a combination of information (for reading), suggested actions (to be applied if a user selects them), and links (to launch an app if the user selects them). The CDS Client decides how to display cards, but this specification recommends displaying suggestions using buttons, and links using underlined text.

https://cds-hooks.org/specification/current/#card-attributes

Source code in healthchain/models/responses/cdsresponse.py

class Card(BaseModel):
    """
    Cards can provide a combination of information (for reading), suggested actions
    (to be applied if a user selects them), and links (to launch an app if the user selects them).
    The CDS Client decides how to display cards, but this specification recommends displaying suggestions
    using buttons, and links using underlined text.

    https://cds-hooks.org/specification/current/#card-attributes
    """

    summary: str = Field(..., max_length=140)
    indicator: IndicatorEnum
    source: Source
    uuid: Optional[str] = None
    detail: Optional[str] = None
    suggestions: Optional[List[Suggestion]] = None
    selectionBehavior: Optional[SelectionBehaviorEnum] = None
    overrideReasons: Optional[List[SimpleCoding]] = None
    links: Optional[List[Link]] = None

    @model_validator(mode="after")
    def validate_suggestions(self) -> Self:
        if self.suggestions is not None:
            assert self.selectionBehavior, f"'selectionBehavior' must be given if 'suggestions' is present! Choose from {[v for v in SelectionBehaviorEnum.value]}"
        return self

IndicatorEnum

Bases: str, Enum

Urgency/importance of what Card conveys. Allowed values, in order of increasing urgency, are: info, warning, critical. The CDS Client MAY use this field to help make UI display decisions such as sort order or coloring.

Source code in healthchain/models/responses/cdsresponse.py

class IndicatorEnum(str, Enum):
    """
    Urgency/importance of what Card conveys.
    Allowed values, in order of increasing urgency, are: info, warning, critical.
    The CDS Client MAY use this field to help make UI display decisions such as sort order or coloring.
    """

    info = "info"
    warning = "warning"
    critical = "critical"

Link

Bases: BaseModel

• CDS Client support for appContext requires additional coordination with the authorization server that is not described or specified in CDS Hooks nor SMART.

• Autolaunchable is experimental

https://cds-hooks.org/specification/current/#link

Source code in healthchain/models/responses/cdsresponse.py

class Link(BaseModel):
    """
    * CDS Client support for appContext requires additional coordination with the authorization
    server that is not described or specified in CDS Hooks nor SMART.

    * Autolaunchable is experimental

    https://cds-hooks.org/specification/current/#link
    """

    label: str
    url: HttpUrl
    type: LinkTypeEnum
    appContext: Optional[str] = None
    autoLaunchable: Optional[bool]

    @model_validator(mode="after")
    def validate_link(self) -> Self:
        if self.appContext:
            assert (
                self.type == LinkTypeEnum.smart
            ), "'type' must be 'smart' for appContext to be valued."

        return self

LinkTypeEnum

Bases: str, Enum

The type of the given URL. There are two possible values for this field. A type of absolute indicates that the URL is absolute and should be treated as-is. A type of smart indicates that the URL is a SMART app launch URL and the CDS Client should ensure the SMART app launch URL is populated with the appropriate SMART launch parameters.

Source code in healthchain/models/responses/cdsresponse.py

class LinkTypeEnum(str, Enum):
    """
    The type of the given URL. There are two possible values for this field.
    A type of absolute indicates that the URL is absolute and should be treated as-is.
    A type of smart indicates that the URL is a SMART app launch URL and the CDS Client
    should ensure the SMART app launch URL is populated with the appropriate SMART
    launch parameters.
    """

    absolute = "absolute"
    smart = "smart"

SelectionBehaviorEnum

Bases: str, Enum

Describes the intended selection behavior of the suggestions in the card. Allowed values are: at-most-one, indicating that the user may choose none or at most one of the suggestions; any, indicating that the end user may choose any number of suggestions including none of them and all of them. CDS Clients that do not understand the value MUST treat the card as an error.

Source code in healthchain/models/responses/cdsresponse.py

class SelectionBehaviorEnum(str, Enum):
    """
    Describes the intended selection behavior of the suggestions in the card.
    Allowed values are: at-most-one, indicating that the user may choose none or
    at most one of the suggestions; any, indicating that the end user may choose
    any number of suggestions including none of them and all of them.
    CDS Clients that do not understand the value MUST treat the card as an error.
    """

    at_most_one = "at-most-one"
    any = "any"

SimpleCoding

Bases: BaseModel

The Coding data type captures the concept of a code. This coding type is a standalone data type in CDS Hooks modeled after a trimmed down version of the FHIR Coding data type.

Source code in healthchain/models/responses/cdsresponse.py

class SimpleCoding(BaseModel):
    """
    The Coding data type captures the concept of a code. This coding type is a standalone data type
    in CDS Hooks modeled after a trimmed down version of the FHIR Coding data type.
    """

    code: str
    system: str
    display: Optional[str] = None

Source

Bases: BaseModel

Grouping structure for the Source of the information displayed on this card. The source should be the primary source of guidance for the decision support Card represents.

https://cds-hooks.org/specification/current/#source

Source code in healthchain/models/responses/cdsresponse.py

class Source(BaseModel):
    """
    Grouping structure for the Source of the information displayed on this card.
    The source should be the primary source of guidance for the decision support Card represents.

    https://cds-hooks.org/specification/current/#source
    """

    label: str
    url: Optional[HttpUrl] = None
    icon: Optional[HttpUrl] = None
    topic: Optional[SimpleCoding] = None

Suggestion

Bases: BaseModel

Allows a service to suggest a set of changes in the context of the current activity (e.g. changing the dose of a medication currently being prescribed, for the order-sign activity). If suggestions are present, selectionBehavior MUST also be provided.

https://cds-hooks.org/specification/current/#suggestion

Source code in healthchain/models/responses/cdsresponse.py

class Suggestion(BaseModel):
    """
    Allows a service to suggest a set of changes in the context of the current activity
    (e.g. changing the dose of a medication currently being prescribed, for the order-sign activity).
    If suggestions are present, selectionBehavior MUST also be provided.

    https://cds-hooks.org/specification/current/#suggestion
    """

    label: str
    uuid: Optional[str] = None
    isRecommended: Optional[bool]
    actions: Optional[List[Action]] = []

CdaRequest

Bases: BaseModel

Source code in healthchain/models/requests/cdarequest.py

class CdaRequest(BaseModel):
    document: str
    session_id: Optional[str] = None
    work_type: Optional[str] = None
    organization_id: Optional[str] = None

    @classmethod
    def from_dict(cls, data: Dict):
        """
        Loads data from dict (xmltodict format)
        """
        return cls(document=xmltodict.unparse(data))

    def model_dump(self, *args, **kwargs) -> Dict:
        """
        Dumps document as dict with xmltodict
        """
        return xmltodict.parse(self.document)

    def model_dump_xml(self, *args, **kwargs) -> str:
        """
        Decodes and dumps document as an xml string
        """
        xml_dict = xmltodict.parse(self.document)
        document = search_key(xml_dict, "urn:Document")
        if document is None:
            log.warning("Couldn't find document under namespace 'urn:Document")
            return ""

        cda = base64.b64decode(document).decode("UTF-8")

        return cda

from_dict(data) classmethod

Loads data from dict (xmltodict format)

Source code in healthchain/models/requests/cdarequest.py

@classmethod
def from_dict(cls, data: Dict):
    """
    Loads data from dict (xmltodict format)
    """
    return cls(document=xmltodict.unparse(data))

model_dump(*args, **kwargs)

Dumps document as dict with xmltodict

Source code in healthchain/models/requests/cdarequest.py

def model_dump(self, *args, **kwargs) -> Dict:
    """
    Dumps document as dict with xmltodict
    """
    return xmltodict.parse(self.document)

model_dump_xml(*args, **kwargs)

Decodes and dumps document as an xml string

Source code in healthchain/models/requests/cdarequest.py

def model_dump_xml(self, *args, **kwargs) -> str:
    """
    Decodes and dumps document as an xml string
    """
    xml_dict = xmltodict.parse(self.document)
    document = search_key(xml_dict, "urn:Document")
    if document is None:
        log.warning("Couldn't find document under namespace 'urn:Document")
        return ""

    cda = base64.b64decode(document).decode("UTF-8")

    return cda

CdaResponse

Bases: BaseModel

Source code in healthchain/models/responses/cdaresponse.py

class CdaResponse(BaseModel):
    document: str
    error: Optional[str] = None

    @classmethod
    def from_dict(cls, data: Dict):
        """
        Loads data from dict (xmltodict format)
        """
        return cls(document=xmltodict.unparse(data))

    def model_dump(self, *args, **kwargs) -> Dict:
        """
        Dumps document as dict with xmltodict
        """
        return xmltodict.parse(self.document)

    def model_dump_xml(self, *args, **kwargs) -> str:
        """
        Decodes and dumps document as an xml string
        """
        xml_dict = xmltodict.parse(self.document)
        document = search_key(xml_dict, "tns:Document")
        if document is None:
            log.warning("Couldn't find document under namespace 'tns:Document")
            return ""

        cda = base64.b64decode(document).decode("UTF-8")

        return cda

from_dict(data) classmethod

Loads data from dict (xmltodict format)

Source code in healthchain/models/responses/cdaresponse.py

@classmethod
def from_dict(cls, data: Dict):
    """
    Loads data from dict (xmltodict format)
    """
    return cls(document=xmltodict.unparse(data))

model_dump(*args, **kwargs)

Dumps document as dict with xmltodict

Source code in healthchain/models/responses/cdaresponse.py

def model_dump(self, *args, **kwargs) -> Dict:
    """
    Dumps document as dict with xmltodict
    """
    return xmltodict.parse(self.document)

model_dump_xml(*args, **kwargs)

Decodes and dumps document as an xml string

Source code in healthchain/models/responses/cdaresponse.py

def model_dump_xml(self, *args, **kwargs) -> str:
    """
    Decodes and dumps document as an xml string
    """
    xml_dict = xmltodict.parse(self.document)
    document = search_key(xml_dict, "tns:Document")
    if document is None:
        log.warning("Couldn't find document under namespace 'tns:Document")
        return ""

    cda = base64.b64decode(document).decode("UTF-8")

    return cda