FHIR Helpers

FHIR utilities for HealthChain.

add_coding_to_codeable_concept(codeable_concept, code, system, display=None)

Add a coding to an existing CodeableConcept.

Useful for adding standardized codes (e.g., SNOMED CT) to resources that already have codes from other systems (e.g., ICD-10).

PARAMETER	DESCRIPTION
codeable_concept	The CodeableConcept to add coding to TYPE: CodeableConcept
code	The code value from the code system TYPE: str
system	The code system URI TYPE: str
display	Optional display text for the code TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
CodeableConcept	The updated CodeableConcept with the new coding added TYPE: CodeableConcept

Example

Add SNOMED CT code to a condition that has ICD-10

condition_code = condition.code condition_code = add_coding_to_codeable_concept( ... condition_code, ... code="44054006", ... system="http://snomed.info/sct", ... display="Type 2 diabetes mellitus" ... )

Source code in healthchain/fhir/helpers.py

def add_coding_to_codeable_concept(
    codeable_concept: CodeableConcept,
    code: str,
    system: str,
    display: Optional[str] = None,
) -> CodeableConcept:
    """Add a coding to an existing CodeableConcept.

    Useful for adding standardized codes (e.g., SNOMED CT) to resources that already
    have codes from other systems (e.g., ICD-10).

    Args:
        codeable_concept: The CodeableConcept to add coding to
        code: The code value from the code system
        system: The code system URI
        display: Optional display text for the code

    Returns:
        CodeableConcept: The updated CodeableConcept with the new coding added

    Example:
        >>> # Add SNOMED CT code to a condition that has ICD-10
        >>> condition_code = condition.code
        >>> condition_code = add_coding_to_codeable_concept(
        ...     condition_code,
        ...     code="44054006",
        ...     system="http://snomed.info/sct",
        ...     display="Type 2 diabetes mellitus"
        ... )
    """
    if not codeable_concept.coding:
        codeable_concept.coding = []

    codeable_concept.coding.append(Coding(system=system, code=code, display=display))

    return codeable_concept

add_provenance_metadata(resource, source, tag_code=None, tag_display=None)

Add provenance metadata to a FHIR resource.

Adds source system identifier, timestamp, and optional processing tags to track data lineage and transformations for audit trails.

PARAMETER	DESCRIPTION
resource	The FHIR resource to annotate TYPE: Resource
source	Name of the source system (e.g., "epic", "cerner") TYPE: str
tag_code	Optional tag code for processing operations (e.g., "aggregated", "deduplicated") TYPE: Optional[str] DEFAULT: None
tag_display	Optional display text for the tag TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
Resource	The resource with added provenance metadata TYPE: Resource

Example

condition = create_condition(subject="Patient/123", code="E11.9") condition = add_provenance_metadata(condition, "epic", "aggregated", "Aggregated from source")

Source code in healthchain/fhir/helpers.py

def add_provenance_metadata(
    resource: Resource,
    source: str,
    tag_code: Optional[str] = None,
    tag_display: Optional[str] = None,
) -> Resource:
    """Add provenance metadata to a FHIR resource.

    Adds source system identifier, timestamp, and optional processing tags to track
    data lineage and transformations for audit trails.

    Args:
        resource: The FHIR resource to annotate
        source: Name of the source system (e.g., "epic", "cerner")
        tag_code: Optional tag code for processing operations (e.g., "aggregated", "deduplicated")
        tag_display: Optional display text for the tag

    Returns:
        Resource: The resource with added provenance metadata

    Example:
        >>> condition = create_condition(subject="Patient/123", code="E11.9")
        >>> condition = add_provenance_metadata(condition, "epic", "aggregated", "Aggregated from source")
    """
    if not resource.meta:
        resource.meta = Meta()

    # Add source system identifier
    resource.meta.source = f"urn:healthchain:source:{source}"

    # Update timestamp
    resource.meta.lastUpdated = datetime.datetime.now(datetime.timezone.utc).isoformat()

    # Add processing tag if provided
    if tag_code:
        if not resource.meta.tag:
            resource.meta.tag = []

        resource.meta.tag.append(
            Coding(
                system="https://dotimplement.github.io/HealthChain/fhir/tags",
                code=tag_code,
                display=tag_display or tag_code,
            )
        )

    return resource

add_resource(bundle, resource, full_url=None)

Add a resource to a bundle.

PARAMETER	DESCRIPTION
bundle	The bundle to add to TYPE: Bundle
resource	The resource to add, e.g. Condition, MedicationStatement, AllergyIntolerance TYPE: Resource
full_url	Optional full URL for the resource TYPE: Optional[str] DEFAULT: None

Source code in healthchain/fhir/bundle_helpers.py

def add_resource(
    bundle: Bundle, resource: Resource, full_url: Optional[str] = None
) -> None:
    """Add a resource to a bundle.

    Args:
        bundle: The bundle to add to
        resource: The resource to add, e.g. Condition, MedicationStatement, AllergyIntolerance
        full_url: Optional full URL for the resource
    """
    entry = BundleEntry(resource=resource)
    if full_url:
        entry.fullUrl = full_url
    bundle.entry = (bundle.entry or []) + [entry]

create_allergy_intolerance(patient, code=None, display=None, system='http://snomed.info/sct')

Create a minimal active FHIR AllergyIntolerance. If you need to create a more complex allergy intolerance, use the FHIR AllergyIntolerance resource directly. https://build.fhir.org/allergyintolerance.html

PARAMETER	DESCRIPTION
patient	REQUIRED. Reference to the patient (e.g. "Patient/123") TYPE: str
code	The allergen code TYPE: Optional[str] DEFAULT: None
display	The display name for the allergen TYPE: Optional[str] DEFAULT: None
system	The code system (default: SNOMED CT) TYPE: Optional[str] DEFAULT: 'http://snomed.info/sct'

RETURNS	DESCRIPTION
AllergyIntolerance	A FHIR AllergyIntolerance resource with an auto-generated ID prefixed with 'hc-' TYPE: AllergyIntolerance

Source code in healthchain/fhir/helpers.py

def create_allergy_intolerance(
    patient: str,
    code: Optional[str] = None,
    display: Optional[str] = None,
    system: Optional[str] = "http://snomed.info/sct",
) -> AllergyIntolerance:
    """
    Create a minimal active FHIR AllergyIntolerance.
    If you need to create a more complex allergy intolerance, use the FHIR AllergyIntolerance resource directly.
    https://build.fhir.org/allergyintolerance.html

    Args:
        patient: REQUIRED. Reference to the patient (e.g. "Patient/123")
        code: The allergen code
        display: The display name for the allergen
        system: The code system (default: SNOMED CT)

    Returns:
        AllergyIntolerance: A FHIR AllergyIntolerance resource with an auto-generated ID prefixed with 'hc-'
    """
    if code:
        allergy_code = create_single_codeable_concept(code, display, system)
    else:
        allergy_code = None

    allergy = AllergyIntolerance(
        id=_generate_id(),
        patient=Reference(reference=patient),
        code=allergy_code,
    )

    return allergy

create_bundle(bundle_type='collection')

Create an empty FHIR Bundle. https://www.hl7.org/fhir/bundle.html

PARAMETER	DESCRIPTION
bundle_type	The type of bundle (default: collection) Valid types: document, message, transaction, transaction-response, batch, batch-response, history, searchset, collection TYPE: str DEFAULT: 'collection'

Source code in healthchain/fhir/bundle_helpers.py

def create_bundle(bundle_type: str = "collection") -> Bundle:
    """Create an empty FHIR Bundle.
    https://www.hl7.org/fhir/bundle.html

    Args:
        bundle_type: The type of bundle (default: collection)
            Valid types: document, message, transaction, transaction-response,
            batch, batch-response, history, searchset, collection
    """
    return Bundle(type=bundle_type, entry=[])

create_condition(subject, clinical_status='active', code=None, display=None, system='http://snomed.info/sct')

Create a minimal active FHIR Condition. If you need to create a more complex condition, use the FHIR Condition resource directly. https://build.fhir.org/condition.html

PARAMETER	DESCRIPTION
subject	REQUIRED. Reference to the patient (e.g. "Patient/123") TYPE: str
clinical_status	REQUIRED. Clinical status (default: active) TYPE: str DEFAULT: 'active'
code	The condition code TYPE: Optional[str] DEFAULT: None
display	The display name for the condition TYPE: Optional[str] DEFAULT: None
system	The code system (default: SNOMED CT) TYPE: Optional[str] DEFAULT: 'http://snomed.info/sct'

RETURNS	DESCRIPTION
Condition	A FHIR Condition resource with an auto-generated ID prefixed with 'hc-' TYPE: Condition

Source code in healthchain/fhir/helpers.py

def create_condition(
    subject: str,
    clinical_status: str = "active",
    code: Optional[str] = None,
    display: Optional[str] = None,
    system: Optional[str] = "http://snomed.info/sct",
) -> Condition:
    """
    Create a minimal active FHIR Condition.
    If you need to create a more complex condition, use the FHIR Condition resource directly.
    https://build.fhir.org/condition.html

    Args:
        subject: REQUIRED. Reference to the patient (e.g. "Patient/123")
        clinical_status: REQUIRED. Clinical status (default: active)
        code: The condition code
        display: The display name for the condition
        system: The code system (default: SNOMED CT)

    Returns:
        Condition: A FHIR Condition resource with an auto-generated ID prefixed with 'hc-'
    """
    if code:
        condition_code = create_single_codeable_concept(code, display, system)
    else:
        condition_code = None

    condition = Condition(
        id=_generate_id(),
        subject=Reference(reference=subject),
        clinicalStatus=create_single_codeable_concept(
            code=clinical_status,
            display=clinical_status.capitalize(),
            system="http://terminology.hl7.org/CodeSystem/condition-clinical",
        ),
        code=condition_code,
    )

    return condition

create_document_reference(data=None, url=None, content_type=None, status='current', description='DocumentReference created by HealthChain', attachment_title='Attachment created by HealthChain')

Create a minimal FHIR DocumentReference. If you need to create a more complex document reference, use the FHIR DocumentReference resource directly. https://build.fhir.org/documentreference.html

PARAMETER	DESCRIPTION
data	The data content of the document attachment TYPE: Optional[Any] DEFAULT: None
url	URL where the document can be accessed TYPE: Optional[str] DEFAULT: None
content_type	MIME type of the document (e.g. "application/pdf", "text/xml", "image/png") TYPE: Optional[str] DEFAULT: None
status	REQUIRED. Status of the document reference (default: current) TYPE: str DEFAULT: 'current'
description	Description of the document reference TYPE: Optional[str] DEFAULT: 'DocumentReference created by HealthChain'
attachment_title	Title for the document attachment TYPE: Optional[str] DEFAULT: 'Attachment created by HealthChain'

RETURNS	DESCRIPTION
DocumentReference	A FHIR DocumentReference resource with an auto-generated ID prefixed with 'hc-' TYPE: DocumentReference

Source code in healthchain/fhir/helpers.py

def create_document_reference(
    data: Optional[Any] = None,
    url: Optional[str] = None,
    content_type: Optional[str] = None,
    status: str = "current",
    description: Optional[str] = "DocumentReference created by HealthChain",
    attachment_title: Optional[str] = "Attachment created by HealthChain",
) -> DocumentReference:
    """
    Create a minimal FHIR DocumentReference.
    If you need to create a more complex document reference, use the FHIR DocumentReference resource directly.
    https://build.fhir.org/documentreference.html

    Args:
        data: The data content of the document attachment
        url: URL where the document can be accessed
        content_type: MIME type of the document (e.g. "application/pdf", "text/xml", "image/png")
        status: REQUIRED. Status of the document reference (default: current)
        description: Description of the document reference
        attachment_title: Title for the document attachment

    Returns:
        DocumentReference: A FHIR DocumentReference resource with an auto-generated ID prefixed with 'hc-'
    """
    document_reference = DocumentReference(
        id=_generate_id(),
        status=status,
        date=datetime.datetime.now(datetime.timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%S%z"
        ),
        description=description,
        content=[
            {
                "attachment": create_single_attachment(
                    content_type=content_type,
                    data=data,
                    url=url,
                    title=attachment_title,
                )
            }
        ],
    )

    return document_reference

create_medication_statement(subject, status='recorded', code=None, display=None, system='http://snomed.info/sct')

Create a minimal recorded FHIR MedicationStatement. If you need to create a more complex medication statement, use the FHIR MedicationStatement resource directly. https://build.fhir.org/medicationstatement.html

PARAMETER	DESCRIPTION
subject	REQUIRED. Reference to the patient (e.g. "Patient/123") TYPE: str
status	REQUIRED. Status of the medication (default: recorded) TYPE: Optional[str] DEFAULT: 'recorded'
code	The medication code TYPE: Optional[str] DEFAULT: None
display	The display name for the medication TYPE: Optional[str] DEFAULT: None
system	The code system (default: SNOMED CT) TYPE: Optional[str] DEFAULT: 'http://snomed.info/sct'

RETURNS	DESCRIPTION
MedicationStatement	A FHIR MedicationStatement resource with an auto-generated ID prefixed with 'hc-' TYPE: MedicationStatement

Source code in healthchain/fhir/helpers.py

def create_medication_statement(
    subject: str,
    status: Optional[str] = "recorded",
    code: Optional[str] = None,
    display: Optional[str] = None,
    system: Optional[str] = "http://snomed.info/sct",
) -> MedicationStatement:
    """
    Create a minimal recorded FHIR MedicationStatement.
    If you need to create a more complex medication statement, use the FHIR MedicationStatement resource directly.
    https://build.fhir.org/medicationstatement.html

    Args:
        subject: REQUIRED. Reference to the patient (e.g. "Patient/123")
        status: REQUIRED. Status of the medication (default: recorded)
        code: The medication code
        display: The display name for the medication
        system: The code system (default: SNOMED CT)

    Returns:
        MedicationStatement: A FHIR MedicationStatement resource with an auto-generated ID prefixed with 'hc-'
    """
    if code:
        medication_concept = create_single_codeable_concept(code, display, system)
    else:
        medication_concept = None

    medication = MedicationStatement(
        id=_generate_id(),
        subject=Reference(reference=subject),
        status=status,
        medication={"concept": medication_concept},
    )

    return medication

create_resource_from_dict(resource_dict, resource_type)

Create a FHIR resource instance from a dictionary

PARAMETER	DESCRIPTION
resource_dict	Dictionary representation of the resource TYPE: Dict
resource_type	Type of FHIR resource to create TYPE: str

RETURNS	DESCRIPTION
Optional[Resource]	Optional[Resource]: FHIR resource instance or None if creation failed

Source code in healthchain/fhir/helpers.py

def create_resource_from_dict(
    resource_dict: Dict, resource_type: str
) -> Optional[Resource]:
    """Create a FHIR resource instance from a dictionary

    Args:
        resource_dict: Dictionary representation of the resource
        resource_type: Type of FHIR resource to create

    Returns:
        Optional[Resource]: FHIR resource instance or None if creation failed
    """
    try:
        resource_module = importlib.import_module(
            f"fhir.resources.{resource_type.lower()}"
        )
        resource_class = getattr(resource_module, resource_type)
        return resource_class(**resource_dict)
    except Exception as e:
        logger.error(f"Failed to create FHIR resource: {str(e)}")
        return None

create_single_attachment(content_type=None, data=None, url=None, title='Attachment created by HealthChain')

Create a minimal FHIR Attachment.

Creates a FHIR Attachment resource with basic fields. Either data or url should be provided. If data is provided, it will be base64 encoded.

PARAMETER	DESCRIPTION
content_type	The MIME type of the content TYPE: Optional[str] DEFAULT: None
data	The actual data content to be base64 encoded TYPE: Optional[str] DEFAULT: None
url	The URL where the data can be found TYPE: Optional[str] DEFAULT: None
title	A title for the attachment (default: "Attachment created by HealthChain") TYPE: Optional[str] DEFAULT: 'Attachment created by HealthChain'

RETURNS	DESCRIPTION
Attachment	A FHIR Attachment resource with basic metadata and content TYPE: Attachment

Source code in healthchain/fhir/helpers.py

def create_single_attachment(
    content_type: Optional[str] = None,
    data: Optional[str] = None,
    url: Optional[str] = None,
    title: Optional[str] = "Attachment created by HealthChain",
) -> Attachment:
    """Create a minimal FHIR Attachment.

    Creates a FHIR Attachment resource with basic fields. Either data or url should be provided.
    If data is provided, it will be base64 encoded.

    Args:
        content_type: The MIME type of the content
        data: The actual data content to be base64 encoded
        url: The URL where the data can be found
        title: A title for the attachment (default: "Attachment created by HealthChain")

    Returns:
        Attachment: A FHIR Attachment resource with basic metadata and content
    """

    if not data and not url:
        logger.warning("No data or url provided for attachment")

    if data:
        data = base64.b64encode(data.encode("utf-8")).decode("utf-8")

    return Attachment(
        contentType=content_type,
        data=data,
        url=url,
        title=title,
        creation=datetime.datetime.now(datetime.timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%S%z"
        ),
    )

create_single_codeable_concept(code, display=None, system='http://snomed.info/sct')

Create a minimal FHIR CodeableConcept with a single coding.

PARAMETER	DESCRIPTION
code	REQUIRED. The code value from the code system TYPE: str
display	The display name for the code TYPE: Optional[str] DEFAULT: None
system	The code system (default: SNOMED CT) TYPE: Optional[str] DEFAULT: 'http://snomed.info/sct'

RETURNS	DESCRIPTION
CodeableConcept	A FHIR CodeableConcept resource with a single coding TYPE: CodeableConcept

Source code in healthchain/fhir/helpers.py

def create_single_codeable_concept(
    code: str,
    display: Optional[str] = None,
    system: Optional[str] = "http://snomed.info/sct",
) -> CodeableConcept:
    """
    Create a minimal FHIR CodeableConcept with a single coding.

    Args:
        code: REQUIRED. The code value from the code system
        display: The display name for the code
        system: The code system (default: SNOMED CT)

    Returns:
        CodeableConcept: A FHIR CodeableConcept resource with a single coding
    """
    return CodeableConcept(coding=[Coding(system=system, code=code, display=display)])

create_single_reaction(code, display=None, system='http://snomed.info/sct', severity=None)

Create a minimal FHIR Reaction with a single coding.

Creates a FHIR Reaction object with a single manifestation coding. The manifestation describes the clinical reaction that was observed. The severity indicates how severe the reaction was.

PARAMETER	DESCRIPTION
code	REQUIRED. The code value from the code system representing the reaction manifestation TYPE: str
display	The display name for the manifestation code TYPE: Optional[str] DEFAULT: None
system	The code system for the manifestation code (default: SNOMED CT) TYPE: Optional[str] DEFAULT: 'http://snomed.info/sct'
severity	The severity of the reaction (mild, moderate, severe) TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict[str, Any]]	A list containing a single FHIR Reaction dictionary with manifestation and severity fields

Source code in healthchain/fhir/helpers.py

def create_single_reaction(
    code: str,
    display: Optional[str] = None,
    system: Optional[str] = "http://snomed.info/sct",
    severity: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """Create a minimal FHIR Reaction with a single coding.

    Creates a FHIR Reaction object with a single manifestation coding. The manifestation
    describes the clinical reaction that was observed. The severity indicates how severe
    the reaction was.

    Args:
        code: REQUIRED. The code value from the code system representing the reaction manifestation
        display: The display name for the manifestation code
        system: The code system for the manifestation code (default: SNOMED CT)
        severity: The severity of the reaction (mild, moderate, severe)

    Returns:
        A list containing a single FHIR Reaction dictionary with manifestation and severity fields
    """
    return [
        {
            "manifestation": [
                CodeableReference(
                    concept=CodeableConcept(
                        coding=[Coding(system=system, code=code, display=display)]
                    )
                )
            ],
            "severity": severity,
        }
    ]

extract_resources(bundle, resource_type)

Remove resources of a given type from a bundle and return them.

Useful for extracting and separating specific resource types (e.g., OperationOutcome) from a FHIR Bundle, modifying the bundle in place.

PARAMETER	DESCRIPTION
bundle	The FHIR Bundle to process (modified in place) TYPE: Bundle
resource_type	The FHIR resource class or string name to extract (e.g., OperationOutcome or "OperationOutcome") TYPE: Union[str, Type[Resource]]

RETURNS	DESCRIPTION
List[Resource]	List[Resource]: All resources of the specified type that were in the bundle

Source code in healthchain/fhir/bundle_helpers.py

def extract_resources(
    bundle: Bundle, resource_type: Union[str, Type[Resource]]
) -> List[Resource]:
    """Remove resources of a given type from a bundle and return them.

    Useful for extracting and separating specific resource types (e.g., OperationOutcome)
    from a FHIR Bundle, modifying the bundle in place.

    Args:
        bundle: The FHIR Bundle to process (modified in place)
        resource_type: The FHIR resource class or string name to extract (e.g., OperationOutcome or "OperationOutcome")

    Returns:
        List[Resource]: All resources of the specified type that were in the bundle
    """
    if not bundle or not bundle.entry:
        return []

    type_class = get_resource_type(resource_type)

    extracted: List[Resource] = []
    remaining_entries: List[BundleEntry] = []

    for entry in bundle.entry:
        resource = entry.resource
        if isinstance(resource, type_class):
            extracted.append(resource)
            continue
        remaining_entries.append(entry)

    bundle.entry = remaining_entries
    return extracted

get_resources(bundle, resource_type)

Get all resources of a specific type from a bundle.

PARAMETER	DESCRIPTION
bundle	The bundle to search TYPE: Bundle
resource_type	String name of the resource type (e.g. "Condition") or the type itself TYPE: Union[str, Type[Resource]]

RETURNS	DESCRIPTION
List[Resource]	List of resources of the specified type

Example

bundle = create_bundle()

Using string identifier

conditions = get_resources(bundle, "Condition") medications = get_resources(bundle, "MedicationStatement") allergies = get_resources(bundle, "AllergyIntolerance")

Or using type directly

from fhir.resources.condition import Condition conditions = get_resources(bundle, Condition)

Source code in healthchain/fhir/bundle_helpers.py

def get_resources(
    bundle: Bundle, resource_type: Union[str, Type[Resource]]
) -> List[Resource]:
    """Get all resources of a specific type from a bundle.

    Args:
        bundle: The bundle to search
        resource_type: String name of the resource type (e.g. "Condition") or the type itself

    Returns:
        List of resources of the specified type

    Example:
        >>> bundle = create_bundle()
        >>> # Using string identifier
        >>> conditions = get_resources(bundle, "Condition")
        >>> medications = get_resources(bundle, "MedicationStatement")
        >>> allergies = get_resources(bundle, "AllergyIntolerance")
        >>>
        >>> # Or using type directly
        >>> from fhir.resources.condition import Condition
        >>> conditions = get_resources(bundle, Condition)
    """
    type_class = get_resource_type(resource_type)
    return [
        entry.resource
        for entry in (bundle.entry or [])
        if isinstance(entry.resource, type_class)
    ]

merge_bundles(bundles, bundle_type='collection', deduplicate=False, dedupe_key='id')

Merge multiple FHIR bundles into a single bundle.

Combines entries from multiple bundles while preserving resource metadata. Useful for aggregating search results from multiple FHIR sources.

PARAMETER	DESCRIPTION
bundles	List of bundles to merge TYPE: List[Bundle]
bundle_type	Type for the merged bundle (default: "collection") TYPE: str DEFAULT: 'collection'
deduplicate	If True, remove duplicate resources based on dedupe_key TYPE: bool DEFAULT: False
dedupe_key	Resource attribute to use for deduplication (default: "id") TYPE: str DEFAULT: 'id'

RETURNS	DESCRIPTION
Bundle	A new bundle containing all entries from input bundles

Example

Merge search results from multiple sources

epic_bundle = gateway.search(Condition, {"patient": "123"}, "epic") cerner_bundle = gateway.search(Condition, {"patient": "123"}, "cerner") merged = merge_bundles([epic_bundle, cerner_bundle], deduplicate=True)

Use in Document workflow

doc = Document(data=merged) doc.fhir.bundle # Contains all conditions from both sources

Source code in healthchain/fhir/bundle_helpers.py

def merge_bundles(
    bundles: List[Bundle],
    bundle_type: str = "collection",
    deduplicate: bool = False,
    dedupe_key: str = "id",
) -> Bundle:
    """Merge multiple FHIR bundles into a single bundle.

    Combines entries from multiple bundles while preserving resource metadata.
    Useful for aggregating search results from multiple FHIR sources.

    Args:
        bundles: List of bundles to merge
        bundle_type: Type for the merged bundle (default: "collection")
        deduplicate: If True, remove duplicate resources based on dedupe_key
        dedupe_key: Resource attribute to use for deduplication (default: "id")

    Returns:
        A new bundle containing all entries from input bundles

    Example:
        >>> # Merge search results from multiple sources
        >>> epic_bundle = gateway.search(Condition, {"patient": "123"}, "epic")
        >>> cerner_bundle = gateway.search(Condition, {"patient": "123"}, "cerner")
        >>> merged = merge_bundles([epic_bundle, cerner_bundle], deduplicate=True)
        >>>
        >>> # Use in Document workflow
        >>> doc = Document(data=merged)
        >>> doc.fhir.bundle  # Contains all conditions from both sources
    """
    merged = create_bundle(bundle_type=bundle_type)

    if deduplicate:
        # Track seen resources by dedupe_key to avoid duplicates
        seen_keys = set()

        for bundle in bundles:
            if not bundle or not bundle.entry:
                continue

            for entry in bundle.entry:
                if not entry.resource:
                    continue

                # Get the deduplication key value
                key_value = getattr(entry.resource, dedupe_key, None)

                # Skip if we've seen this key before
                if key_value and key_value in seen_keys:
                    continue

                # Add to merged bundle and track the key
                add_resource(merged, entry.resource, entry.fullUrl)
                if key_value:
                    seen_keys.add(key_value)
    else:
        # No deduplication - just merge all entries
        for bundle in bundles:
            if not bundle or not bundle.entry:
                continue

            for entry in bundle.entry:
                if entry.resource:
                    add_resource(merged, entry.resource, entry.fullUrl)

    return merged

read_content_attachment(document_reference, include_data=True)

Read the attachments in a human readable format from a FHIR DocumentReference content field.

PARAMETER	DESCRIPTION
document_reference	The FHIR DocumentReference resource TYPE: DocumentReference
include_data	Whether to include the data of the attachments. If true, the data will be also be decoded (default: True) TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
Optional[List[Dict[str, Any]]]	Optional[List[Dict[str, Any]]]: List of dictionaries containing attachment data and metadata, or None if no attachments are found: [ { "data": str, "metadata": Dict[str, Any] } ]

Source code in healthchain/fhir/helpers.py

def read_content_attachment(
    document_reference: DocumentReference,
    include_data: bool = True,
) -> Optional[List[Dict[str, Any]]]:
    """Read the attachments in a human readable format from a FHIR DocumentReference content field.

    Args:
        document_reference: The FHIR DocumentReference resource
        include_data: Whether to include the data of the attachments. If true, the data will be also be decoded (default: True)

    Returns:
        Optional[List[Dict[str, Any]]]: List of dictionaries containing attachment data and metadata,
            or None if no attachments are found:
            [
                {
                    "data": str,
                    "metadata": Dict[str, Any]
                }
            ]
    """
    if not document_reference.content:
        return None

    attachments = []
    for content in document_reference.content:
        attachment = content.attachment
        result = {}

        if include_data:
            result["data"] = (
                attachment.url if attachment.url else attachment.data.decode("utf-8")
            )

        result["metadata"] = {
            "content_type": attachment.contentType,
            "title": attachment.title,
            "creation": attachment.creation,
        }

        attachments.append(result)

    return attachments

set_condition_category(condition, category)

Set the category of a FHIR Condition to either 'problem-list-item' or 'encounter-diagnosis'.

PARAMETER	DESCRIPTION
condition	The FHIR Condition resource to modify TYPE: Condition
category	The category to set. Must be 'problem-list-item' or 'encounter-diagnosis'. TYPE: str

RETURNS	DESCRIPTION
Condition	The modified FHIR Condition resource with the specified category set TYPE: Condition

RAISES	DESCRIPTION
ValueError	If the category is not one of the allowed values.

Source code in healthchain/fhir/helpers.py

def set_condition_category(condition: Condition, category: str) -> Condition:
    """
    Set the category of a FHIR Condition to either 'problem-list-item' or 'encounter-diagnosis'.

    Args:
        condition: The FHIR Condition resource to modify
        category: The category to set. Must be 'problem-list-item' or 'encounter-diagnosis'.

    Returns:
        Condition: The modified FHIR Condition resource with the specified category set

    Raises:
        ValueError: If the category is not one of the allowed values.
    """
    allowed_categories = {
        "problem-list-item": {
            "code": "problem-list-item",
            "display": "Problem List Item",
        },
        "encounter-diagnosis": {
            "code": "encounter-diagnosis",
            "display": "Encounter Diagnosis",
        },
    }
    if category not in allowed_categories:
        raise ValueError(
            f"Invalid category '{category}'. Must be one of: {list(allowed_categories.keys())}"
        )

    cat_info = allowed_categories[category]
    condition.category = [
        create_single_codeable_concept(
            code=cat_info["code"],
            display=cat_info["display"],
            system="http://terminology.hl7.org/CodeSystem/condition-category",
        )
    ]
    return condition

set_resources(bundle, resources, resource_type, replace=True)

Set resources of a specific type in the bundle.

PARAMETER	DESCRIPTION
bundle	The bundle to modify TYPE: Bundle
resources	The new resources to add TYPE: List[Resource]
resource_type	String name of the resource type (e.g. "Condition") or the type itself TYPE: Union[str, Type[Resource]]
replace	If True, remove existing resources of this type before adding new ones. If False, append new resources to existing ones. Defaults to True. TYPE: bool DEFAULT: True

Example

bundle = create_bundle()

Append to existing resources (default behavior)

set_resources(bundle, [condition1, condition2], "Condition") set_resources(bundle, [medication1], "MedicationStatement")

Replace existing resources

set_resources(bundle, [condition3], "Condition", replace=True)

Or using type directly

from fhir.resources.condition import Condition set_resources(bundle, [condition1, condition2], Condition)

Source code in healthchain/fhir/bundle_helpers.py

def set_resources(
    bundle: Bundle,
    resources: List[Resource],
    resource_type: Union[str, Type[Resource]],
    replace: bool = True,
) -> None:
    """Set resources of a specific type in the bundle.

    Args:
        bundle: The bundle to modify
        resources: The new resources to add
        resource_type: String name of the resource type (e.g. "Condition") or the type itself
        replace: If True, remove existing resources of this type before adding new ones.
                If False, append new resources to existing ones. Defaults to True.

    Example:
        >>> bundle = create_bundle()
        >>> # Append to existing resources (default behavior)
        >>> set_resources(bundle, [condition1, condition2], "Condition")
        >>> set_resources(bundle, [medication1], "MedicationStatement")
        >>>
        >>> # Replace existing resources
        >>> set_resources(bundle, [condition3], "Condition", replace=True)
        >>>
        >>> # Or using type directly
        >>> from fhir.resources.condition import Condition
        >>> set_resources(bundle, [condition1, condition2], Condition)
    """
    type_class = get_resource_type(resource_type)

    # Remove existing resources of this type if replace=True
    if replace:
        bundle.entry = [
            entry
            for entry in (bundle.entry or [])
            if not isinstance(entry.resource, type_class)
        ]

    # Add new resources
    for resource in resources:
        if not isinstance(resource, type_class):
            raise ValueError(
                f"Resource must be of type {type_class.__name__}, "
                f"got {type(resource).__name__}"
            )
        add_resource(bundle, resource)

bundle_helpers

Helper functions for working with FHIR Bundles. Patterns: - create_(): create a new FHIR bundle - add_(): add a resource to a bundle - get_(): get resources from a bundle - set_(): set resources in a bundle - merge_(): merge multiple bundles into a single bundle - extract_(): extract resources from a bundle

add_resource(bundle, resource, full_url=None)

Add a resource to a bundle.

PARAMETER	DESCRIPTION
bundle	The bundle to add to TYPE: Bundle
resource	The resource to add, e.g. Condition, MedicationStatement, AllergyIntolerance TYPE: Resource
full_url	Optional full URL for the resource TYPE: Optional[str] DEFAULT: None

Source code in healthchain/fhir/bundle_helpers.py

def add_resource(
    bundle: Bundle, resource: Resource, full_url: Optional[str] = None
) -> None:
    """Add a resource to a bundle.

    Args:
        bundle: The bundle to add to
        resource: The resource to add, e.g. Condition, MedicationStatement, AllergyIntolerance
        full_url: Optional full URL for the resource
    """
    entry = BundleEntry(resource=resource)
    if full_url:
        entry.fullUrl = full_url
    bundle.entry = (bundle.entry or []) + [entry]

create_bundle(bundle_type='collection')

Create an empty FHIR Bundle. https://www.hl7.org/fhir/bundle.html

PARAMETER	DESCRIPTION
bundle_type	The type of bundle (default: collection) Valid types: document, message, transaction, transaction-response, batch, batch-response, history, searchset, collection TYPE: str DEFAULT: 'collection'

Source code in healthchain/fhir/bundle_helpers.py

def create_bundle(bundle_type: str = "collection") -> Bundle:
    """Create an empty FHIR Bundle.
    https://www.hl7.org/fhir/bundle.html

    Args:
        bundle_type: The type of bundle (default: collection)
            Valid types: document, message, transaction, transaction-response,
            batch, batch-response, history, searchset, collection
    """
    return Bundle(type=bundle_type, entry=[])

extract_resources(bundle, resource_type)

Remove resources of a given type from a bundle and return them.

Useful for extracting and separating specific resource types (e.g., OperationOutcome) from a FHIR Bundle, modifying the bundle in place.

PARAMETER	DESCRIPTION
bundle	The FHIR Bundle to process (modified in place) TYPE: Bundle
resource_type	The FHIR resource class or string name to extract (e.g., OperationOutcome or "OperationOutcome") TYPE: Union[str, Type[Resource]]

RETURNS	DESCRIPTION
List[Resource]	List[Resource]: All resources of the specified type that were in the bundle

Source code in healthchain/fhir/bundle_helpers.py

def extract_resources(
    bundle: Bundle, resource_type: Union[str, Type[Resource]]
) -> List[Resource]:
    """Remove resources of a given type from a bundle and return them.

    Useful for extracting and separating specific resource types (e.g., OperationOutcome)
    from a FHIR Bundle, modifying the bundle in place.

    Args:
        bundle: The FHIR Bundle to process (modified in place)
        resource_type: The FHIR resource class or string name to extract (e.g., OperationOutcome or "OperationOutcome")

    Returns:
        List[Resource]: All resources of the specified type that were in the bundle
    """
    if not bundle or not bundle.entry:
        return []

    type_class = get_resource_type(resource_type)

    extracted: List[Resource] = []
    remaining_entries: List[BundleEntry] = []

    for entry in bundle.entry:
        resource = entry.resource
        if isinstance(resource, type_class):
            extracted.append(resource)
            continue
        remaining_entries.append(entry)

    bundle.entry = remaining_entries
    return extracted

get_resource_type(resource_type)

Get the resource type class from string or type.

PARAMETER	DESCRIPTION
resource_type	String name of the resource type (e.g. "Condition") or the type itself TYPE: Union[str, Type[Resource]]

RETURNS	DESCRIPTION
Type[Resource]	The resource type class

RAISES	DESCRIPTION
ValueError	If the resource type is not supported or cannot be imported

Source code in healthchain/fhir/bundle_helpers.py

def get_resource_type(resource_type: Union[str, Type[Resource]]) -> Type[Resource]:
    """Get the resource type class from string or type.

    Args:
        resource_type: String name of the resource type (e.g. "Condition") or the type itself

    Returns:
        The resource type class

    Raises:
        ValueError: If the resource type is not supported or cannot be imported
    """
    if isinstance(resource_type, type) and issubclass(resource_type, Resource):
        return resource_type

    if not isinstance(resource_type, str):
        raise ValueError(
            f"Resource type must be a string or Resource class, got {type(resource_type)}"
        )

    try:
        # Try to import the resource type dynamically from fhir.resources
        module = __import__(
            f"fhir.resources.{resource_type.lower()}", fromlist=[resource_type]
        )
        return getattr(module, resource_type)
    except (ImportError, AttributeError) as e:
        raise ValueError(
            f"Could not import resource type: {resource_type}. "
            "Make sure it is a valid FHIR resource type."
        ) from e

get_resources(bundle, resource_type)

Get all resources of a specific type from a bundle.

PARAMETER	DESCRIPTION
bundle	The bundle to search TYPE: Bundle
resource_type	String name of the resource type (e.g. "Condition") or the type itself TYPE: Union[str, Type[Resource]]

RETURNS	DESCRIPTION
List[Resource]	List of resources of the specified type

Example

bundle = create_bundle()

Using string identifier

conditions = get_resources(bundle, "Condition") medications = get_resources(bundle, "MedicationStatement") allergies = get_resources(bundle, "AllergyIntolerance")

Or using type directly

from fhir.resources.condition import Condition conditions = get_resources(bundle, Condition)

Source code in healthchain/fhir/bundle_helpers.py

def get_resources(
    bundle: Bundle, resource_type: Union[str, Type[Resource]]
) -> List[Resource]:
    """Get all resources of a specific type from a bundle.

    Args:
        bundle: The bundle to search
        resource_type: String name of the resource type (e.g. "Condition") or the type itself

    Returns:
        List of resources of the specified type

    Example:
        >>> bundle = create_bundle()
        >>> # Using string identifier
        >>> conditions = get_resources(bundle, "Condition")
        >>> medications = get_resources(bundle, "MedicationStatement")
        >>> allergies = get_resources(bundle, "AllergyIntolerance")
        >>>
        >>> # Or using type directly
        >>> from fhir.resources.condition import Condition
        >>> conditions = get_resources(bundle, Condition)
    """
    type_class = get_resource_type(resource_type)
    return [
        entry.resource
        for entry in (bundle.entry or [])
        if isinstance(entry.resource, type_class)
    ]

merge_bundles(bundles, bundle_type='collection', deduplicate=False, dedupe_key='id')

Merge multiple FHIR bundles into a single bundle.

Combines entries from multiple bundles while preserving resource metadata. Useful for aggregating search results from multiple FHIR sources.

PARAMETER	DESCRIPTION
bundles	List of bundles to merge TYPE: List[Bundle]
bundle_type	Type for the merged bundle (default: "collection") TYPE: str DEFAULT: 'collection'
deduplicate	If True, remove duplicate resources based on dedupe_key TYPE: bool DEFAULT: False
dedupe_key	Resource attribute to use for deduplication (default: "id") TYPE: str DEFAULT: 'id'

RETURNS	DESCRIPTION
Bundle	A new bundle containing all entries from input bundles

Example

Merge search results from multiple sources

epic_bundle = gateway.search(Condition, {"patient": "123"}, "epic") cerner_bundle = gateway.search(Condition, {"patient": "123"}, "cerner") merged = merge_bundles([epic_bundle, cerner_bundle], deduplicate=True)

Use in Document workflow

doc = Document(data=merged) doc.fhir.bundle # Contains all conditions from both sources

Source code in healthchain/fhir/bundle_helpers.py

def merge_bundles(
    bundles: List[Bundle],
    bundle_type: str = "collection",
    deduplicate: bool = False,
    dedupe_key: str = "id",
) -> Bundle:
    """Merge multiple FHIR bundles into a single bundle.

    Combines entries from multiple bundles while preserving resource metadata.
    Useful for aggregating search results from multiple FHIR sources.

    Args:
        bundles: List of bundles to merge
        bundle_type: Type for the merged bundle (default: "collection")
        deduplicate: If True, remove duplicate resources based on dedupe_key
        dedupe_key: Resource attribute to use for deduplication (default: "id")

    Returns:
        A new bundle containing all entries from input bundles

    Example:
        >>> # Merge search results from multiple sources
        >>> epic_bundle = gateway.search(Condition, {"patient": "123"}, "epic")
        >>> cerner_bundle = gateway.search(Condition, {"patient": "123"}, "cerner")
        >>> merged = merge_bundles([epic_bundle, cerner_bundle], deduplicate=True)
        >>>
        >>> # Use in Document workflow
        >>> doc = Document(data=merged)
        >>> doc.fhir.bundle  # Contains all conditions from both sources
    """
    merged = create_bundle(bundle_type=bundle_type)

    if deduplicate:
        # Track seen resources by dedupe_key to avoid duplicates
        seen_keys = set()

        for bundle in bundles:
            if not bundle or not bundle.entry:
                continue

            for entry in bundle.entry:
                if not entry.resource:
                    continue

                # Get the deduplication key value
                key_value = getattr(entry.resource, dedupe_key, None)

                # Skip if we've seen this key before
                if key_value and key_value in seen_keys:
                    continue

                # Add to merged bundle and track the key
                add_resource(merged, entry.resource, entry.fullUrl)
                if key_value:
                    seen_keys.add(key_value)
    else:
        # No deduplication - just merge all entries
        for bundle in bundles:
            if not bundle or not bundle.entry:
                continue

            for entry in bundle.entry:
                if entry.resource:
                    add_resource(merged, entry.resource, entry.fullUrl)

    return merged

set_resources(bundle, resources, resource_type, replace=True)

Set resources of a specific type in the bundle.

PARAMETER	DESCRIPTION
bundle	The bundle to modify TYPE: Bundle
resources	The new resources to add TYPE: List[Resource]
resource_type	String name of the resource type (e.g. "Condition") or the type itself TYPE: Union[str, Type[Resource]]
replace	If True, remove existing resources of this type before adding new ones. If False, append new resources to existing ones. Defaults to True. TYPE: bool DEFAULT: True

Example

bundle = create_bundle()

Append to existing resources (default behavior)

set_resources(bundle, [condition1, condition2], "Condition") set_resources(bundle, [medication1], "MedicationStatement")

Replace existing resources

set_resources(bundle, [condition3], "Condition", replace=True)

Or using type directly

from fhir.resources.condition import Condition set_resources(bundle, [condition1, condition2], Condition)

Source code in healthchain/fhir/bundle_helpers.py

def set_resources(
    bundle: Bundle,
    resources: List[Resource],
    resource_type: Union[str, Type[Resource]],
    replace: bool = True,
) -> None:
    """Set resources of a specific type in the bundle.

    Args:
        bundle: The bundle to modify
        resources: The new resources to add
        resource_type: String name of the resource type (e.g. "Condition") or the type itself
        replace: If True, remove existing resources of this type before adding new ones.
                If False, append new resources to existing ones. Defaults to True.

    Example:
        >>> bundle = create_bundle()
        >>> # Append to existing resources (default behavior)
        >>> set_resources(bundle, [condition1, condition2], "Condition")
        >>> set_resources(bundle, [medication1], "MedicationStatement")
        >>>
        >>> # Replace existing resources
        >>> set_resources(bundle, [condition3], "Condition", replace=True)
        >>>
        >>> # Or using type directly
        >>> from fhir.resources.condition import Condition
        >>> set_resources(bundle, [condition1, condition2], Condition)
    """
    type_class = get_resource_type(resource_type)

    # Remove existing resources of this type if replace=True
    if replace:
        bundle.entry = [
            entry
            for entry in (bundle.entry or [])
            if not isinstance(entry.resource, type_class)
        ]

    # Add new resources
    for resource in resources:
        if not isinstance(resource, type_class):
            raise ValueError(
                f"Resource must be of type {type_class.__name__}, "
                f"got {type(resource).__name__}"
            )
        add_resource(bundle, resource)

helpers

Convenience functions for creating minimal FHIR resources. Patterns: - create_(): create a new FHIR resource with sensible defaults - useful for dev, use with caution - add_(): add data to resources with list fields safely (e.g. coding) - set_(): set the field of specific resources with soft validation (e.g. category) - read_(): return a human readable format of the data in a resource (e.g. attachments)

add_coding_to_codeable_concept(codeable_concept, code, system, display=None)

Add a coding to an existing CodeableConcept.

Useful for adding standardized codes (e.g., SNOMED CT) to resources that already have codes from other systems (e.g., ICD-10).

PARAMETER	DESCRIPTION
codeable_concept	The CodeableConcept to add coding to TYPE: CodeableConcept
code	The code value from the code system TYPE: str
system	The code system URI TYPE: str
display	Optional display text for the code TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
CodeableConcept	The updated CodeableConcept with the new coding added TYPE: CodeableConcept

Example

Add SNOMED CT code to a condition that has ICD-10

condition_code = condition.code condition_code = add_coding_to_codeable_concept( ... condition_code, ... code="44054006", ... system="http://snomed.info/sct", ... display="Type 2 diabetes mellitus" ... )

Source code in healthchain/fhir/helpers.py

def add_coding_to_codeable_concept(
    codeable_concept: CodeableConcept,
    code: str,
    system: str,
    display: Optional[str] = None,
) -> CodeableConcept:
    """Add a coding to an existing CodeableConcept.

    Useful for adding standardized codes (e.g., SNOMED CT) to resources that already
    have codes from other systems (e.g., ICD-10).

    Args:
        codeable_concept: The CodeableConcept to add coding to
        code: The code value from the code system
        system: The code system URI
        display: Optional display text for the code

    Returns:
        CodeableConcept: The updated CodeableConcept with the new coding added

    Example:
        >>> # Add SNOMED CT code to a condition that has ICD-10
        >>> condition_code = condition.code
        >>> condition_code = add_coding_to_codeable_concept(
        ...     condition_code,
        ...     code="44054006",
        ...     system="http://snomed.info/sct",
        ...     display="Type 2 diabetes mellitus"
        ... )
    """
    if not codeable_concept.coding:
        codeable_concept.coding = []

    codeable_concept.coding.append(Coding(system=system, code=code, display=display))

    return codeable_concept

add_provenance_metadata(resource, source, tag_code=None, tag_display=None)

Add provenance metadata to a FHIR resource.

Adds source system identifier, timestamp, and optional processing tags to track data lineage and transformations for audit trails.

PARAMETER	DESCRIPTION
resource	The FHIR resource to annotate TYPE: Resource
source	Name of the source system (e.g., "epic", "cerner") TYPE: str
tag_code	Optional tag code for processing operations (e.g., "aggregated", "deduplicated") TYPE: Optional[str] DEFAULT: None
tag_display	Optional display text for the tag TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
Resource	The resource with added provenance metadata TYPE: Resource

Example

condition = create_condition(subject="Patient/123", code="E11.9") condition = add_provenance_metadata(condition, "epic", "aggregated", "Aggregated from source")

Source code in healthchain/fhir/helpers.py

def add_provenance_metadata(
    resource: Resource,
    source: str,
    tag_code: Optional[str] = None,
    tag_display: Optional[str] = None,
) -> Resource:
    """Add provenance metadata to a FHIR resource.

    Adds source system identifier, timestamp, and optional processing tags to track
    data lineage and transformations for audit trails.

    Args:
        resource: The FHIR resource to annotate
        source: Name of the source system (e.g., "epic", "cerner")
        tag_code: Optional tag code for processing operations (e.g., "aggregated", "deduplicated")
        tag_display: Optional display text for the tag

    Returns:
        Resource: The resource with added provenance metadata

    Example:
        >>> condition = create_condition(subject="Patient/123", code="E11.9")
        >>> condition = add_provenance_metadata(condition, "epic", "aggregated", "Aggregated from source")
    """
    if not resource.meta:
        resource.meta = Meta()

    # Add source system identifier
    resource.meta.source = f"urn:healthchain:source:{source}"

    # Update timestamp
    resource.meta.lastUpdated = datetime.datetime.now(datetime.timezone.utc).isoformat()

    # Add processing tag if provided
    if tag_code:
        if not resource.meta.tag:
            resource.meta.tag = []

        resource.meta.tag.append(
            Coding(
                system="https://dotimplement.github.io/HealthChain/fhir/tags",
                code=tag_code,
                display=tag_display or tag_code,
            )
        )

    return resource

create_allergy_intolerance(patient, code=None, display=None, system='http://snomed.info/sct')

Create a minimal active FHIR AllergyIntolerance. If you need to create a more complex allergy intolerance, use the FHIR AllergyIntolerance resource directly. https://build.fhir.org/allergyintolerance.html

PARAMETER	DESCRIPTION
patient	REQUIRED. Reference to the patient (e.g. "Patient/123") TYPE: str
code	The allergen code TYPE: Optional[str] DEFAULT: None
display	The display name for the allergen TYPE: Optional[str] DEFAULT: None
system	The code system (default: SNOMED CT) TYPE: Optional[str] DEFAULT: 'http://snomed.info/sct'

RETURNS	DESCRIPTION
AllergyIntolerance	A FHIR AllergyIntolerance resource with an auto-generated ID prefixed with 'hc-' TYPE: AllergyIntolerance

Source code in healthchain/fhir/helpers.py

def create_allergy_intolerance(
    patient: str,
    code: Optional[str] = None,
    display: Optional[str] = None,
    system: Optional[str] = "http://snomed.info/sct",
) -> AllergyIntolerance:
    """
    Create a minimal active FHIR AllergyIntolerance.
    If you need to create a more complex allergy intolerance, use the FHIR AllergyIntolerance resource directly.
    https://build.fhir.org/allergyintolerance.html

    Args:
        patient: REQUIRED. Reference to the patient (e.g. "Patient/123")
        code: The allergen code
        display: The display name for the allergen
        system: The code system (default: SNOMED CT)

    Returns:
        AllergyIntolerance: A FHIR AllergyIntolerance resource with an auto-generated ID prefixed with 'hc-'
    """
    if code:
        allergy_code = create_single_codeable_concept(code, display, system)
    else:
        allergy_code = None

    allergy = AllergyIntolerance(
        id=_generate_id(),
        patient=Reference(reference=patient),
        code=allergy_code,
    )

    return allergy

create_condition(subject, clinical_status='active', code=None, display=None, system='http://snomed.info/sct')

Create a minimal active FHIR Condition. If you need to create a more complex condition, use the FHIR Condition resource directly. https://build.fhir.org/condition.html

PARAMETER	DESCRIPTION
subject	REQUIRED. Reference to the patient (e.g. "Patient/123") TYPE: str
clinical_status	REQUIRED. Clinical status (default: active) TYPE: str DEFAULT: 'active'
code	The condition code TYPE: Optional[str] DEFAULT: None
display	The display name for the condition TYPE: Optional[str] DEFAULT: None
system	The code system (default: SNOMED CT) TYPE: Optional[str] DEFAULT: 'http://snomed.info/sct'

RETURNS	DESCRIPTION
Condition	A FHIR Condition resource with an auto-generated ID prefixed with 'hc-' TYPE: Condition

Source code in healthchain/fhir/helpers.py

def create_condition(
    subject: str,
    clinical_status: str = "active",
    code: Optional[str] = None,
    display: Optional[str] = None,
    system: Optional[str] = "http://snomed.info/sct",
) -> Condition:
    """
    Create a minimal active FHIR Condition.
    If you need to create a more complex condition, use the FHIR Condition resource directly.
    https://build.fhir.org/condition.html

    Args:
        subject: REQUIRED. Reference to the patient (e.g. "Patient/123")
        clinical_status: REQUIRED. Clinical status (default: active)
        code: The condition code
        display: The display name for the condition
        system: The code system (default: SNOMED CT)

    Returns:
        Condition: A FHIR Condition resource with an auto-generated ID prefixed with 'hc-'
    """
    if code:
        condition_code = create_single_codeable_concept(code, display, system)
    else:
        condition_code = None

    condition = Condition(
        id=_generate_id(),
        subject=Reference(reference=subject),
        clinicalStatus=create_single_codeable_concept(
            code=clinical_status,
            display=clinical_status.capitalize(),
            system="http://terminology.hl7.org/CodeSystem/condition-clinical",
        ),
        code=condition_code,
    )

    return condition

create_document_reference(data=None, url=None, content_type=None, status='current', description='DocumentReference created by HealthChain', attachment_title='Attachment created by HealthChain')

Create a minimal FHIR DocumentReference. If you need to create a more complex document reference, use the FHIR DocumentReference resource directly. https://build.fhir.org/documentreference.html

PARAMETER	DESCRIPTION
data	The data content of the document attachment TYPE: Optional[Any] DEFAULT: None
url	URL where the document can be accessed TYPE: Optional[str] DEFAULT: None
content_type	MIME type of the document (e.g. "application/pdf", "text/xml", "image/png") TYPE: Optional[str] DEFAULT: None
status	REQUIRED. Status of the document reference (default: current) TYPE: str DEFAULT: 'current'
description	Description of the document reference TYPE: Optional[str] DEFAULT: 'DocumentReference created by HealthChain'
attachment_title	Title for the document attachment TYPE: Optional[str] DEFAULT: 'Attachment created by HealthChain'

RETURNS	DESCRIPTION
DocumentReference	A FHIR DocumentReference resource with an auto-generated ID prefixed with 'hc-' TYPE: DocumentReference

Source code in healthchain/fhir/helpers.py

def create_document_reference(
    data: Optional[Any] = None,
    url: Optional[str] = None,
    content_type: Optional[str] = None,
    status: str = "current",
    description: Optional[str] = "DocumentReference created by HealthChain",
    attachment_title: Optional[str] = "Attachment created by HealthChain",
) -> DocumentReference:
    """
    Create a minimal FHIR DocumentReference.
    If you need to create a more complex document reference, use the FHIR DocumentReference resource directly.
    https://build.fhir.org/documentreference.html

    Args:
        data: The data content of the document attachment
        url: URL where the document can be accessed
        content_type: MIME type of the document (e.g. "application/pdf", "text/xml", "image/png")
        status: REQUIRED. Status of the document reference (default: current)
        description: Description of the document reference
        attachment_title: Title for the document attachment

    Returns:
        DocumentReference: A FHIR DocumentReference resource with an auto-generated ID prefixed with 'hc-'
    """
    document_reference = DocumentReference(
        id=_generate_id(),
        status=status,
        date=datetime.datetime.now(datetime.timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%S%z"
        ),
        description=description,
        content=[
            {
                "attachment": create_single_attachment(
                    content_type=content_type,
                    data=data,
                    url=url,
                    title=attachment_title,
                )
            }
        ],
    )

    return document_reference

create_medication_statement(subject, status='recorded', code=None, display=None, system='http://snomed.info/sct')

Create a minimal recorded FHIR MedicationStatement. If you need to create a more complex medication statement, use the FHIR MedicationStatement resource directly. https://build.fhir.org/medicationstatement.html

PARAMETER	DESCRIPTION
subject	REQUIRED. Reference to the patient (e.g. "Patient/123") TYPE: str
status	REQUIRED. Status of the medication (default: recorded) TYPE: Optional[str] DEFAULT: 'recorded'
code	The medication code TYPE: Optional[str] DEFAULT: None
display	The display name for the medication TYPE: Optional[str] DEFAULT: None
system	The code system (default: SNOMED CT) TYPE: Optional[str] DEFAULT: 'http://snomed.info/sct'

RETURNS	DESCRIPTION
MedicationStatement	A FHIR MedicationStatement resource with an auto-generated ID prefixed with 'hc-' TYPE: MedicationStatement

Source code in healthchain/fhir/helpers.py

def create_medication_statement(
    subject: str,
    status: Optional[str] = "recorded",
    code: Optional[str] = None,
    display: Optional[str] = None,
    system: Optional[str] = "http://snomed.info/sct",
) -> MedicationStatement:
    """
    Create a minimal recorded FHIR MedicationStatement.
    If you need to create a more complex medication statement, use the FHIR MedicationStatement resource directly.
    https://build.fhir.org/medicationstatement.html

    Args:
        subject: REQUIRED. Reference to the patient (e.g. "Patient/123")
        status: REQUIRED. Status of the medication (default: recorded)
        code: The medication code
        display: The display name for the medication
        system: The code system (default: SNOMED CT)

    Returns:
        MedicationStatement: A FHIR MedicationStatement resource with an auto-generated ID prefixed with 'hc-'
    """
    if code:
        medication_concept = create_single_codeable_concept(code, display, system)
    else:
        medication_concept = None

    medication = MedicationStatement(
        id=_generate_id(),
        subject=Reference(reference=subject),
        status=status,
        medication={"concept": medication_concept},
    )

    return medication

create_resource_from_dict(resource_dict, resource_type)

Create a FHIR resource instance from a dictionary

PARAMETER	DESCRIPTION
resource_dict	Dictionary representation of the resource TYPE: Dict
resource_type	Type of FHIR resource to create TYPE: str

RETURNS	DESCRIPTION
Optional[Resource]	Optional[Resource]: FHIR resource instance or None if creation failed

Source code in healthchain/fhir/helpers.py

def create_resource_from_dict(
    resource_dict: Dict, resource_type: str
) -> Optional[Resource]:
    """Create a FHIR resource instance from a dictionary

    Args:
        resource_dict: Dictionary representation of the resource
        resource_type: Type of FHIR resource to create

    Returns:
        Optional[Resource]: FHIR resource instance or None if creation failed
    """
    try:
        resource_module = importlib.import_module(
            f"fhir.resources.{resource_type.lower()}"
        )
        resource_class = getattr(resource_module, resource_type)
        return resource_class(**resource_dict)
    except Exception as e:
        logger.error(f"Failed to create FHIR resource: {str(e)}")
        return None

create_single_attachment(content_type=None, data=None, url=None, title='Attachment created by HealthChain')

Create a minimal FHIR Attachment.

Creates a FHIR Attachment resource with basic fields. Either data or url should be provided. If data is provided, it will be base64 encoded.

PARAMETER	DESCRIPTION
content_type	The MIME type of the content TYPE: Optional[str] DEFAULT: None
data	The actual data content to be base64 encoded TYPE: Optional[str] DEFAULT: None
url	The URL where the data can be found TYPE: Optional[str] DEFAULT: None
title	A title for the attachment (default: "Attachment created by HealthChain") TYPE: Optional[str] DEFAULT: 'Attachment created by HealthChain'

RETURNS	DESCRIPTION
Attachment	A FHIR Attachment resource with basic metadata and content TYPE: Attachment

Source code in healthchain/fhir/helpers.py

def create_single_attachment(
    content_type: Optional[str] = None,
    data: Optional[str] = None,
    url: Optional[str] = None,
    title: Optional[str] = "Attachment created by HealthChain",
) -> Attachment:
    """Create a minimal FHIR Attachment.

    Creates a FHIR Attachment resource with basic fields. Either data or url should be provided.
    If data is provided, it will be base64 encoded.

    Args:
        content_type: The MIME type of the content
        data: The actual data content to be base64 encoded
        url: The URL where the data can be found
        title: A title for the attachment (default: "Attachment created by HealthChain")

    Returns:
        Attachment: A FHIR Attachment resource with basic metadata and content
    """

    if not data and not url:
        logger.warning("No data or url provided for attachment")

    if data:
        data = base64.b64encode(data.encode("utf-8")).decode("utf-8")

    return Attachment(
        contentType=content_type,
        data=data,
        url=url,
        title=title,
        creation=datetime.datetime.now(datetime.timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%S%z"
        ),
    )

create_single_codeable_concept(code, display=None, system='http://snomed.info/sct')

Create a minimal FHIR CodeableConcept with a single coding.

PARAMETER	DESCRIPTION
code	REQUIRED. The code value from the code system TYPE: str
display	The display name for the code TYPE: Optional[str] DEFAULT: None
system	The code system (default: SNOMED CT) TYPE: Optional[str] DEFAULT: 'http://snomed.info/sct'

RETURNS	DESCRIPTION
CodeableConcept	A FHIR CodeableConcept resource with a single coding TYPE: CodeableConcept

Source code in healthchain/fhir/helpers.py

def create_single_codeable_concept(
    code: str,
    display: Optional[str] = None,
    system: Optional[str] = "http://snomed.info/sct",
) -> CodeableConcept:
    """
    Create a minimal FHIR CodeableConcept with a single coding.

    Args:
        code: REQUIRED. The code value from the code system
        display: The display name for the code
        system: The code system (default: SNOMED CT)

    Returns:
        CodeableConcept: A FHIR CodeableConcept resource with a single coding
    """
    return CodeableConcept(coding=[Coding(system=system, code=code, display=display)])

create_single_reaction(code, display=None, system='http://snomed.info/sct', severity=None)

Create a minimal FHIR Reaction with a single coding.

Creates a FHIR Reaction object with a single manifestation coding. The manifestation describes the clinical reaction that was observed. The severity indicates how severe the reaction was.

PARAMETER	DESCRIPTION
code	REQUIRED. The code value from the code system representing the reaction manifestation TYPE: str
display	The display name for the manifestation code TYPE: Optional[str] DEFAULT: None
system	The code system for the manifestation code (default: SNOMED CT) TYPE: Optional[str] DEFAULT: 'http://snomed.info/sct'
severity	The severity of the reaction (mild, moderate, severe) TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict[str, Any]]	A list containing a single FHIR Reaction dictionary with manifestation and severity fields

Source code in healthchain/fhir/helpers.py

def create_single_reaction(
    code: str,
    display: Optional[str] = None,
    system: Optional[str] = "http://snomed.info/sct",
    severity: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """Create a minimal FHIR Reaction with a single coding.

    Creates a FHIR Reaction object with a single manifestation coding. The manifestation
    describes the clinical reaction that was observed. The severity indicates how severe
    the reaction was.

    Args:
        code: REQUIRED. The code value from the code system representing the reaction manifestation
        display: The display name for the manifestation code
        system: The code system for the manifestation code (default: SNOMED CT)
        severity: The severity of the reaction (mild, moderate, severe)

    Returns:
        A list containing a single FHIR Reaction dictionary with manifestation and severity fields
    """
    return [
        {
            "manifestation": [
                CodeableReference(
                    concept=CodeableConcept(
                        coding=[Coding(system=system, code=code, display=display)]
                    )
                )
            ],
            "severity": severity,
        }
    ]

read_content_attachment(document_reference, include_data=True)

Read the attachments in a human readable format from a FHIR DocumentReference content field.

PARAMETER	DESCRIPTION
document_reference	The FHIR DocumentReference resource TYPE: DocumentReference
include_data	Whether to include the data of the attachments. If true, the data will be also be decoded (default: True) TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
Optional[List[Dict[str, Any]]]	Optional[List[Dict[str, Any]]]: List of dictionaries containing attachment data and metadata, or None if no attachments are found: [ { "data": str, "metadata": Dict[str, Any] } ]

Source code in healthchain/fhir/helpers.py

def read_content_attachment(
    document_reference: DocumentReference,
    include_data: bool = True,
) -> Optional[List[Dict[str, Any]]]:
    """Read the attachments in a human readable format from a FHIR DocumentReference content field.

    Args:
        document_reference: The FHIR DocumentReference resource
        include_data: Whether to include the data of the attachments. If true, the data will be also be decoded (default: True)

    Returns:
        Optional[List[Dict[str, Any]]]: List of dictionaries containing attachment data and metadata,
            or None if no attachments are found:
            [
                {
                    "data": str,
                    "metadata": Dict[str, Any]
                }
            ]
    """
    if not document_reference.content:
        return None

    attachments = []
    for content in document_reference.content:
        attachment = content.attachment
        result = {}

        if include_data:
            result["data"] = (
                attachment.url if attachment.url else attachment.data.decode("utf-8")
            )

        result["metadata"] = {
            "content_type": attachment.contentType,
            "title": attachment.title,
            "creation": attachment.creation,
        }

        attachments.append(result)

    return attachments

set_condition_category(condition, category)

Set the category of a FHIR Condition to either 'problem-list-item' or 'encounter-diagnosis'.

PARAMETER	DESCRIPTION
condition	The FHIR Condition resource to modify TYPE: Condition
category	The category to set. Must be 'problem-list-item' or 'encounter-diagnosis'. TYPE: str

RETURNS	DESCRIPTION
Condition	The modified FHIR Condition resource with the specified category set TYPE: Condition

RAISES	DESCRIPTION
ValueError	If the category is not one of the allowed values.

Source code in healthchain/fhir/helpers.py

def set_condition_category(condition: Condition, category: str) -> Condition:
    """
    Set the category of a FHIR Condition to either 'problem-list-item' or 'encounter-diagnosis'.

    Args:
        condition: The FHIR Condition resource to modify
        category: The category to set. Must be 'problem-list-item' or 'encounter-diagnosis'.

    Returns:
        Condition: The modified FHIR Condition resource with the specified category set

    Raises:
        ValueError: If the category is not one of the allowed values.
    """
    allowed_categories = {
        "problem-list-item": {
            "code": "problem-list-item",
            "display": "Problem List Item",
        },
        "encounter-diagnosis": {
            "code": "encounter-diagnosis",
            "display": "Encounter Diagnosis",
        },
    }
    if category not in allowed_categories:
        raise ValueError(
            f"Invalid category '{category}'. Must be one of: {list(allowed_categories.keys())}"
        )

    cat_info = allowed_categories[category]
    condition.category = [
        create_single_codeable_concept(
            code=cat_info["code"],
            display=cat_info["display"],
            system="http://terminology.hl7.org/CodeSystem/condition-category",
        )
    ]
    return condition