Skip to content

Use Cases

CdsRequestConstructor

Bases: BaseRequestConstructor

Handles the request construction and validation

Source code in healthchain/sandbox/use_cases/cds.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class CdsRequestConstructor(BaseRequestConstructor):
    """
    Handles the request construction and validation
    """

    def __init__(self) -> None:
        self.api_protocol = ApiProtocol.rest
        self.context_mapping = {
            Workflow.order_select: OrderSelectContext,
            Workflow.order_sign: OrderSignContext,
            Workflow.patient_view: PatientViewContext,
            Workflow.encounter_discharge: EncounterDischargeContext,
        }

    @validate_workflow(UseCaseMapping.ClinicalDecisionSupport)
    def construct_request(
        self,
        prefetch_data: Dict[str, Resource],
        workflow: Workflow,
        context: Optional[Dict[str, str]] = {},
    ) -> CDSRequest:
        """
        Constructs a HL7-compliant CDS request with prefetch data.

        Parameters:
            prefetch_data (Dict[str, Resource]): Dictionary mapping prefetch keys to FHIR resources
            workflow (Workflow): The CDS hook name, e.g. patient-view
            context (Optional[Dict[str, str]]): Optional context data for the CDS hook

        Returns:
            CDSRequest: A Pydantic model that wraps a CDS request for REST API

        Raises:
            ValueError: If the workflow is invalid or not implemented
            TypeError: If any prefetch value is not a valid FHIR resource

        # TODO: Add FhirServer support
        """

        log.debug(f"Constructing CDS request for {workflow.value} from {prefetch_data}")

        context_model = self.context_mapping.get(workflow, None)
        if context_model is None:
            raise ValueError(
                f"Invalid workflow {workflow.value} or workflow model not implemented."
            )
        if not isinstance(prefetch_data, Prefetch):
            raise TypeError(
                f"Prefetch data must be a Prefetch object, but got {type(prefetch_data)}"
            )
        request = CDSRequest(
            hook=workflow.value,
            context=context_model(**context),
            prefetch=prefetch_data.prefetch,
        )
        return request

construct_request(prefetch_data, workflow, context={})

Constructs a HL7-compliant CDS request with prefetch data.

PARAMETER DESCRIPTION
prefetch_data

Dictionary mapping prefetch keys to FHIR resources

TYPE: Dict[str, Resource]

workflow

The CDS hook name, e.g. patient-view

TYPE: Workflow

context

Optional context data for the CDS hook

TYPE: Optional[Dict[str, str]] DEFAULT: {}

RETURNS DESCRIPTION
CDSRequest

A Pydantic model that wraps a CDS request for REST API

TYPE: CDSRequest

RAISES DESCRIPTION
ValueError

If the workflow is invalid or not implemented
TypeError

If any prefetch value is not a valid FHIR resource

TODO: Add FhirServer support

Source code in healthchain/sandbox/use_cases/cds.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@validate_workflow(UseCaseMapping.ClinicalDecisionSupport)
def construct_request(
    self,
    prefetch_data: Dict[str, Resource],
    workflow: Workflow,
    context: Optional[Dict[str, str]] = {},
) -> CDSRequest:
    """
    Constructs a HL7-compliant CDS request with prefetch data.

    Parameters:
        prefetch_data (Dict[str, Resource]): Dictionary mapping prefetch keys to FHIR resources
        workflow (Workflow): The CDS hook name, e.g. patient-view
        context (Optional[Dict[str, str]]): Optional context data for the CDS hook

    Returns:
        CDSRequest: A Pydantic model that wraps a CDS request for REST API

    Raises:
        ValueError: If the workflow is invalid or not implemented
        TypeError: If any prefetch value is not a valid FHIR resource

    # TODO: Add FhirServer support
    """

    log.debug(f"Constructing CDS request for {workflow.value} from {prefetch_data}")

    context_model = self.context_mapping.get(workflow, None)
    if context_model is None:
        raise ValueError(
            f"Invalid workflow {workflow.value} or workflow model not implemented."
        )
    if not isinstance(prefetch_data, Prefetch):
        raise TypeError(
            f"Prefetch data must be a Prefetch object, but got {type(prefetch_data)}"
        )
    request = CDSRequest(
        hook=workflow.value,
        context=context_model(**context),
        prefetch=prefetch_data.prefetch,
    )
    return request

ClinicalDecisionSupport

Bases: BaseUseCase

Implements EHR backend simulator for Clinical Decision Support (CDS).

This class provides functionality to simulate CDS Hooks interactions between an EHR system and a CDS service. It handles the construction and sending of CDS Hook requests according to the HL7 CDS Hooks specification.

PARAMETER DESCRIPTION
path

The API endpoint path for CDS services

TYPE: str DEFAULT: '/cds-services/'

client

The client used to send requests to the CDS service

TYPE: Optional[BaseClient] DEFAULT: None

The class uses a CdsRequestConstructor strategy to build properly formatted CDS Hook requests with appropriate context and prefetch data.

See https://cds-hooks.org/ for the complete specification

Source code in healthchain/sandbox/use_cases/cds.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class ClinicalDecisionSupport(BaseUseCase):
    """
    Implements EHR backend simulator for Clinical Decision Support (CDS).

    This class provides functionality to simulate CDS Hooks interactions between
    an EHR system and a CDS service. It handles the construction and sending of
    CDS Hook requests according to the HL7 CDS Hooks specification.

    Parameters:
        path (str): The API endpoint path for CDS services
        client (Optional[BaseClient]): The client used to send requests to the CDS service

    The class uses a CdsRequestConstructor strategy to build properly formatted
    CDS Hook requests with appropriate context and prefetch data.

    See https://cds-hooks.org/ for the complete specification
    """

    def __init__(
        self,
        path: str = "/cds-services/",
        client: Optional[BaseClient] = None,
    ) -> None:
        super().__init__(
            client=client,
        )
        self._type = UseCaseType.cds
        self._strategy = CdsRequestConstructor()
        self._path = path

    @property
    def description(self) -> str:
        return "Clinical decision support (HL7 CDS specification)"

    @property
    def type(self) -> UseCaseType:
        return self._type

    @property
    def strategy(self) -> BaseRequestConstructor:
        return self._strategy

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 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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  # fhir resource is passed either thru prefetched template of fhir server
    )
    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 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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 
33
34
35
36
37
38
39
40
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 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
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 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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 
 8
 9
10
11
12
13
14
15
16
17
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"

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 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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 
43
44
45
46
47
48
49
50
51
52
53
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 
20
21
22
23
24
25
26
27
28
29
30
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 
82
83
84
85
86
87
88
89
90
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 
137
138
139
140
141
142
143
144
145
146
147
148
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 
122
123
124
125
126
127
128
129
130
131
132
133
134
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]] = []

ClinDocRequestConstructor

Bases: BaseRequestConstructor

Handles the request construction and validation of a NoteReader CDA file

Source code in healthchain/sandbox/use_cases/clindoc.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class ClinDocRequestConstructor(BaseRequestConstructor):
    """
    Handles the request construction and validation of a NoteReader CDA file
    """

    def __init__(self) -> None:
        self.api_protocol: ApiProtocol = ApiProtocol.soap
        self.soap_envelope: Dict = self._load_soap_envelope()

    def _load_soap_envelope(self):
        data = pkgutil.get_data("healthchain", "templates/soap_envelope.xml")
        return xmltodict.parse(data.decode("utf-8"))

    def construct_cda_xml_document(self):
        """
        This function should wrap FHIR resources from Document into a template CDA file
        TODO: implement this function
        """
        raise NotImplementedError("This function is not implemented yet.")

    @validate_workflow(UseCaseMapping.ClinicalDocumentation)
    def construct_request(
        self, document_reference: DocumentReference, workflow: Workflow
    ) -> CdaRequest:
        """
        Constructs a CDA request for clinical documentation use cases (NoteReader)

        Parameters:
            document_reference (DocumentReference): FHIR DocumentReference containing CDA XML data
            workflow (Workflow): The NoteReader workflow type, e.g. notereader-sign-inpatient

        Returns:
            CdaRequest: A Pydantic model containing the CDA XML wrapped in a SOAP envelope

        Raises:
            ValueError: If the SOAP envelope template is invalid or missing required keys
        """
        # TODO: handle different workflows
        cda_xml = None
        for content in document_reference.content:
            if content.attachment.contentType == "text/xml":
                cda_xml = content.attachment.data
                break

        if cda_xml is not None:
            # Make a copy of the SOAP envelope template
            soap_envelope = self.soap_envelope.copy()

            cda_xml = base64.b64encode(cda_xml).decode("utf-8")

            # Insert encoded cda in the Document section
            if not insert_at_key(soap_envelope, "urn:Document", cda_xml):
                raise ValueError(
                    "Key 'urn:Document' missing from SOAP envelope template!"
                )
            request = CdaRequest.from_dict(soap_envelope)

            return request
        else:
            log.warning("No CDA document found in the DocumentReference!")

construct_cda_xml_document()

This function should wrap FHIR resources from Document into a template CDA file TODO: implement this function

Source code in healthchain/sandbox/use_cases/clindoc.py 
37
38
39
40
41
42
def construct_cda_xml_document(self):
    """
    This function should wrap FHIR resources from Document into a template CDA file
    TODO: implement this function
    """
    raise NotImplementedError("This function is not implemented yet.")

construct_request(document_reference, workflow)

Constructs a CDA request for clinical documentation use cases (NoteReader)

PARAMETER DESCRIPTION
document_reference

FHIR DocumentReference containing CDA XML data

TYPE: DocumentReference

workflow

The NoteReader workflow type, e.g. notereader-sign-inpatient

TYPE: Workflow

RETURNS DESCRIPTION
CdaRequest

A Pydantic model containing the CDA XML wrapped in a SOAP envelope

TYPE: CdaRequest

RAISES DESCRIPTION
ValueError

If the SOAP envelope template is invalid or missing required keys
Source code in healthchain/sandbox/use_cases/clindoc.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@validate_workflow(UseCaseMapping.ClinicalDocumentation)
def construct_request(
    self, document_reference: DocumentReference, workflow: Workflow
) -> CdaRequest:
    """
    Constructs a CDA request for clinical documentation use cases (NoteReader)

    Parameters:
        document_reference (DocumentReference): FHIR DocumentReference containing CDA XML data
        workflow (Workflow): The NoteReader workflow type, e.g. notereader-sign-inpatient

    Returns:
        CdaRequest: A Pydantic model containing the CDA XML wrapped in a SOAP envelope

    Raises:
        ValueError: If the SOAP envelope template is invalid or missing required keys
    """
    # TODO: handle different workflows
    cda_xml = None
    for content in document_reference.content:
        if content.attachment.contentType == "text/xml":
            cda_xml = content.attachment.data
            break

    if cda_xml is not None:
        # Make a copy of the SOAP envelope template
        soap_envelope = self.soap_envelope.copy()

        cda_xml = base64.b64encode(cda_xml).decode("utf-8")

        # Insert encoded cda in the Document section
        if not insert_at_key(soap_envelope, "urn:Document", cda_xml):
            raise ValueError(
                "Key 'urn:Document' missing from SOAP envelope template!"
            )
        request = CdaRequest.from_dict(soap_envelope)

        return request
    else:
        log.warning("No CDA document found in the DocumentReference!")

ClinicalDocumentation

Bases: BaseUseCase

Implements EHR backend strategy for clinical documentation (NoteReader)

This class represents the backend strategy for clinical documentation using the NoteReader system. It inherits from the BaseUseCase class and provides methods for processing NoteReader documents. When used with the @sandbox decorator, it enables testing and validation of clinical documentation workflows in a controlled environment.

ATTRIBUTE DESCRIPTION
client

The client to be used for communication with the service.

TYPE: Optional[BaseClient]

path

The endpoint path to send requests to. Defaults to "/notereader/". Will be normalized to ensure it starts and ends with a forward slash.

TYPE: str

type

The type of use case, set to UseCaseType.clindoc.

TYPE: UseCaseType

strategy

The strategy used for constructing requests.

TYPE: BaseRequestConstructor

Example

@sandbox("http://localhost:8000") class MyNoteReader(ClinicalDocumentation): def init(self): super().init(path="/custom/notereader/")

Create instance and start sandbox

note_reader = MyNoteReader() note_reader.start_sandbox(save_data=True)

Source code in healthchain/sandbox/use_cases/clindoc.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class ClinicalDocumentation(BaseUseCase):
    """
    Implements EHR backend strategy for clinical documentation (NoteReader)

    This class represents the backend strategy for clinical documentation using the NoteReader system.
    It inherits from the `BaseUseCase` class and provides methods for processing NoteReader documents.
    When used with the @sandbox decorator, it enables testing and validation of clinical documentation
    workflows in a controlled environment.

    Attributes:
        client (Optional[BaseClient]): The client to be used for communication with the service.
        path (str): The endpoint path to send requests to. Defaults to "/notereader/".
                    Will be normalized to ensure it starts and ends with a forward slash.
        type (UseCaseType): The type of use case, set to UseCaseType.clindoc.
        strategy (BaseRequestConstructor): The strategy used for constructing requests.

    Example:
        @sandbox("http://localhost:8000")
        class MyNoteReader(ClinicalDocumentation):
            def __init__(self):
                super().__init__(path="/custom/notereader/")

        # Create instance and start sandbox
        note_reader = MyNoteReader()
        note_reader.start_sandbox(save_data=True)
    """

    def __init__(
        self,
        path: str = "/notereader/",
        client: Optional[BaseClient] = None,
    ) -> None:
        super().__init__(
            client=client,
        )
        self._type = UseCaseType.clindoc
        self._strategy = ClinDocRequestConstructor()
        self._path = path

    @property
    def description(self) -> str:
        return "Clinical documentation (NoteReader)"

    @property
    def type(self) -> UseCaseType:
        return self._type

    @property
    def strategy(self) -> BaseRequestConstructor:
        return self._strategy

CdaRequest

Bases: BaseModel

Source code in healthchain/models/requests/cdarequest.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
19
20
21
22
23
24
@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 
26
27
28
29
30
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 
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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 
18
19
20
21
22
23
@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 
25
26
27
28
29
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 
31
32
33
34
35
36
37
38
39
40
41
42
43
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