FHIR Helpers

FHIR utilities for HealthChain.

BundleConverterConfig

Bases: BaseModel

Configuration for FHIR Bundle to DataFrame conversion.

This configuration object controls which FHIR resources are processed and how they are converted to DataFrame columns for ML model deployment.

ATTRIBUTE	DESCRIPTION
resources	List of FHIR resource types to include in the conversion TYPE: List[str]
observation_aggregation	How to aggregate multiple observation values TYPE: Literal['mean', 'median', 'max', 'min', 'last']
age_calculation	Method for calculating patient age TYPE: Literal['current_date', 'event_date']
event_date_source	Which resource to extract event date from TYPE: Literal['Observation', 'Encounter']
event_date_strategy	Which date to use when multiple dates exist TYPE: Literal['earliest', 'latest', 'first']
resource_options	Resource-specific configuration options (extensible) TYPE: Dict[str, Dict[str, Any]]

Example

config = BundleConverterConfig( ... resources=["Patient", "Observation", "Condition"], ... observation_aggregation="median" ... ) df = bundle_to_dataframe(bundle, config=config)

Source code in healthchain/fhir/dataframe.py

class BundleConverterConfig(BaseModel):
    """Configuration for FHIR Bundle to DataFrame conversion.

    This configuration object controls which FHIR resources are processed and how
    they are converted to DataFrame columns for ML model deployment.

    Attributes:
        resources: List of FHIR resource types to include in the conversion
        observation_aggregation: How to aggregate multiple observation values
        age_calculation: Method for calculating patient age
        event_date_source: Which resource to extract event date from
        event_date_strategy: Which date to use when multiple dates exist
        resource_options: Resource-specific configuration options (extensible)

    Example:
        >>> config = BundleConverterConfig(
        ...     resources=["Patient", "Observation", "Condition"],
        ...     observation_aggregation="median"
        ... )
        >>> df = bundle_to_dataframe(bundle, config=config)
    """

    # Core resources to include
    resources: List[str] = ["Patient", "Observation"]

    # Observation-specific options
    observation_aggregation: Literal["mean", "median", "max", "min", "last"] = "mean"

    # Patient age calculation
    age_calculation: Literal["current_date", "event_date"] = "current_date"
    event_date_source: Literal["Observation", "Encounter"] = "Observation"
    event_date_strategy: Literal["earliest", "latest", "first"] = "earliest"

    # Resource-specific options (extensible for future use)
    resource_options: Dict[str, Dict[str, Any]] = {}

    model_config = ConfigDict(extra="allow")

    @field_validator("resources")
    @classmethod
    def validate_resources(cls, v):
        """Validate that requested resources are supported and warn about unsupported ones."""
        supported = get_supported_resources()
        unsupported = [r for r in v if r not in supported]
        if unsupported:
            logger.warning(
                f"Unsupported resources will be skipped: {unsupported}. "
                f"Supported resources: {supported}"
            )
        return v

validate_resources(v) classmethod

Validate that requested resources are supported and warn about unsupported ones.

Source code in healthchain/fhir/dataframe.py

@field_validator("resources")
@classmethod
def validate_resources(cls, v):
    """Validate that requested resources are supported and warn about unsupported ones."""
    supported = get_supported_resources()
    unsupported = [r for r in v if r not in supported]
    if unsupported:
        logger.warning(
            f"Unsupported resources will be skipped: {unsupported}. "
            f"Supported resources: {supported}"
        )
    return v

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/resourcehelpers.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/resourcehelpers.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/bundlehelpers.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]

bundle_to_dataframe(bundle, config=None)

Convert a FHIR Bundle to a pandas DataFrame.

Converts FHIR resources to a tabular format with one row per patient. Uses a configuration object to control which resources are processed and how.

PARAMETER	DESCRIPTION
bundle	FHIR Bundle resource (object or dict) TYPE: Union[Bundle, Dict[str, Any]]
config	BundleConverterConfig object specifying conversion behavior. If None, uses default config (Patient + Observation with mean aggregation) TYPE: Optional[BundleConverterConfig] DEFAULT: None

RETURNS	DESCRIPTION
DataFrame	DataFrame with one row per patient and columns for each feature

Example

from healthchain.fhir.converters import BundleConverterConfig

Default behavior

df = bundle_to_dataframe(bundle)

Custom config

config = BundleConverterConfig( ... resources=["Patient", "Observation", "Condition"], ... observation_aggregation="median", ... age_calculation="event_date" ... ) df = bundle_to_dataframe(bundle, config=config)

Source code in healthchain/fhir/dataframe.py

def bundle_to_dataframe(
    bundle: Union[Bundle, Dict[str, Any]],
    config: Optional[BundleConverterConfig] = None,
) -> pd.DataFrame:
    """Convert a FHIR Bundle to a pandas DataFrame.

    Converts FHIR resources to a tabular format with one row per patient.
    Uses a configuration object to control which resources are processed and how.

    Args:
        bundle: FHIR Bundle resource (object or dict)
        config: BundleConverterConfig object specifying conversion behavior.
            If None, uses default config (Patient + Observation with mean aggregation)

    Returns:
        DataFrame with one row per patient and columns for each feature

    Example:
        >>> from healthchain.fhir.converters import BundleConverterConfig
        >>>
        >>> # Default behavior
        >>> df = bundle_to_dataframe(bundle)
        >>>
        >>> # Custom config
        >>> config = BundleConverterConfig(
        ...     resources=["Patient", "Observation", "Condition"],
        ...     observation_aggregation="median",
        ...     age_calculation="event_date"
        ... )
        >>> df = bundle_to_dataframe(bundle, config=config)
    """
    # Use default config if not provided
    if config is None:
        config = BundleConverterConfig()

    # Group resources by patient
    patient_data = group_bundle_by_patient(bundle)

    if not patient_data:
        return pd.DataFrame()

    # Build rows for each patient
    rows = []
    for patient_ref, resources in patient_data.items():
        row = {"patient_ref": patient_ref}

        # Process each requested resource type using registry
        for resource_type in config.resources:
            handler_info = SUPPORTED_RESOURCES.get(resource_type)

            if not handler_info:
                # Skip unsupported resources gracefully (already warned by validator)
                continue

            # Get handler function by name
            handler_name = handler_info["handler"]
            handler = globals()[handler_name]

            # Call handler with standardized signature
            features = handler(resources, config)
            if features:
                row.update(features)

        rows.append(row)

    return pd.DataFrame(rows)

calculate_age_from_birthdate(birth_date)

Calculate age in years from a birth date string.

PARAMETER	DESCRIPTION
birth_date	Birth date in ISO format (YYYY-MM-DD or full ISO datetime) TYPE: str

RETURNS	DESCRIPTION
Optional[int]	Age in years, or None if birth date is invalid

Source code in healthchain/fhir/utilities.py

def calculate_age_from_birthdate(birth_date: str) -> Optional[int]:
    """Calculate age in years from a birth date string.

    Args:
        birth_date: Birth date in ISO format (YYYY-MM-DD or full ISO datetime)

    Returns:
        Age in years, or None if birth date is invalid
    """
    if not birth_date:
        return None

    try:
        if isinstance(birth_date, str):
            # Remove timezone info for simpler parsing
            birth_date_clean = birth_date.replace("Z", "").split("T")[0]
            birth_dt = datetime.datetime.strptime(birth_date_clean, "%Y-%m-%d")
        else:
            birth_dt = birth_date

        # Calculate age
        today = datetime.datetime.now()
        age = today.year - birth_dt.year

        # Adjust if birthday hasn't occurred this year
        if (today.month, today.day) < (birth_dt.month, birth_dt.day):
            age -= 1

        return age
    except (ValueError, AttributeError, TypeError):
        return None

calculate_age_from_event_date(birth_date, event_date)

Calculate age in years from birth date and event date (MIMIC-IV style).

Uses the formula: age = year(eventDate) - year(birthDate) This matches MIMIC-IV on FHIR de-identified age calculation.

PARAMETER	DESCRIPTION
birth_date	Birth date in ISO format (YYYY-MM-DD or full ISO datetime) TYPE: str
event_date	Event date in ISO format (YYYY-MM-DD or full ISO datetime) TYPE: str

RETURNS	DESCRIPTION
Optional[int]	Age in years based on year difference, or None if dates are invalid

Example

calculate_age_from_event_date("1990-06-15", "2020-03-10") 30

Source code in healthchain/fhir/utilities.py

def calculate_age_from_event_date(birth_date: str, event_date: str) -> Optional[int]:
    """Calculate age in years from birth date and event date (MIMIC-IV style).

    Uses the formula: age = year(eventDate) - year(birthDate)
    This matches MIMIC-IV on FHIR de-identified age calculation.

    Args:
        birth_date: Birth date in ISO format (YYYY-MM-DD or full ISO datetime)
        event_date: Event date in ISO format (YYYY-MM-DD or full ISO datetime)

    Returns:
        Age in years based on year difference, or None if dates are invalid

    Example:
        >>> calculate_age_from_event_date("1990-06-15", "2020-03-10")
        30
    """
    if not birth_date or not event_date:
        return None

    try:
        # Parse birth date
        if isinstance(birth_date, str):
            birth_date_clean = birth_date.replace("Z", "").split("T")[0]
            birth_year = int(birth_date_clean.split("-")[0])
        else:
            birth_year = birth_date.year

        # Parse event date
        if isinstance(event_date, str):
            event_date_clean = event_date.replace("Z", "").split("T")[0]
            event_year = int(event_date_clean.split("-")[0])
        else:
            event_year = event_date.year

        # MIMIC-IV style: simple year difference
        age = event_year - birth_year

        return age
    except (ValueError, AttributeError, TypeError, IndexError):
        return None

convert_prefetch_to_fhir_objects(prefetch_dict)

Convert a dictionary of FHIR resource dicts to FHIR Resource objects.

Takes a prefetch dictionary where values may be either dict representations of FHIR resources or already instantiated FHIR Resource objects, and ensures all values are FHIR Resource objects.

PARAMETER	DESCRIPTION
prefetch_dict	Dictionary mapping keys to FHIR resource dicts or objects TYPE: Dict[str, Any]

RETURNS	DESCRIPTION
Dict[str, Resource]	Dict[str, Resource]: Dictionary with same keys but all values as FHIR Resource objects

Example

prefetch = { ... "patient": {"resourceType": "Patient", "id": "123"}, ... "condition": Condition(id="456", ...) ... } fhir_objects = convert_prefetch_to_fhir_objects(prefetch) isinstance(fhir_objects["patient"], Patient) # True isinstance(fhir_objects["condition"], Condition) # True

Source code in healthchain/fhir/readers.py

def convert_prefetch_to_fhir_objects(
    prefetch_dict: Dict[str, Any],
) -> Dict[str, Resource]:
    """Convert a dictionary of FHIR resource dicts to FHIR Resource objects.

    Takes a prefetch dictionary where values may be either dict representations of FHIR
    resources or already instantiated FHIR Resource objects, and ensures all values are
    FHIR Resource objects.

    Args:
        prefetch_dict: Dictionary mapping keys to FHIR resource dicts or objects

    Returns:
        Dict[str, Resource]: Dictionary with same keys but all values as FHIR Resource objects

    Example:
        >>> prefetch = {
        ...     "patient": {"resourceType": "Patient", "id": "123"},
        ...     "condition": Condition(id="456", ...)
        ... }
        >>> fhir_objects = convert_prefetch_to_fhir_objects(prefetch)
        >>> isinstance(fhir_objects["patient"], Patient)  # True
        >>> isinstance(fhir_objects["condition"], Condition)  # True
    """
    from fhir.resources import get_fhir_model_class

    result: Dict[str, Resource] = {}

    for key, resource_data in prefetch_dict.items():
        if isinstance(resource_data, dict):
            # Convert dict to FHIR Resource object
            resource_type = resource_data.get("resourceType")
            if resource_type:
                try:
                    resource_class = get_fhir_model_class(resource_type)
                    result[key] = resource_class(**resource_data)
                except Exception as e:
                    logger.warning(
                        f"Failed to convert {resource_type} to FHIR object: {e}"
                    )
                    result[key] = resource_data
            else:
                logger.warning(
                    f"No resourceType found for key '{key}', keeping as dict"
                )
                result[key] = resource_data
        elif isinstance(resource_data, Resource):
            # Already a FHIR object
            result[key] = resource_data
        else:
            logger.warning(f"Unexpected type for key '{key}': {type(resource_data)}")
            result[key] = resource_data

    return result

count_resources(bundle)

Count resources by type in a bundle.

PARAMETER	DESCRIPTION
bundle	The FHIR Bundle to analyze TYPE: Bundle

RETURNS	DESCRIPTION
dict[str, int]	Dictionary mapping resource type names to their counts.
Example	{"Condition": 2, "MedicationStatement": 1, "Patient": 1} TYPE: dict[str, int]

Example

bundle = create_bundle() add_resource(bundle, create_condition(...)) add_resource(bundle, create_condition(...)) add_resource(bundle, create_medication_statement(...)) counts = count_resources(bundle) print(counts)

Source code in healthchain/fhir/bundlehelpers.py

def count_resources(bundle: Bundle) -> dict[str, int]:
    """Count resources by type in a bundle.

    Args:
        bundle: The FHIR Bundle to analyze

    Returns:
        Dictionary mapping resource type names to their counts.
        Example: {"Condition": 2, "MedicationStatement": 1, "Patient": 1}

    Example:
        >>> bundle = create_bundle()
        >>> add_resource(bundle, create_condition(...))
        >>> add_resource(bundle, create_condition(...))
        >>> add_resource(bundle, create_medication_statement(...))
        >>> counts = count_resources(bundle)
        >>> print(counts)
        {'Condition': 2, 'MedicationStatement': 1}
    """
    if not bundle or not bundle.entry:
        return {}

    counts: dict[str, int] = {}
    for entry in bundle.entry:
        if entry.resource:
            # Get the resource type from the class name
            resource_type = entry.resource.__resource_type__
            counts[resource_type] = counts.get(resource_type, 0) + 1

    return counts

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/resourcehelpers.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/bundlehelpers.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/resourcehelpers.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/resourcehelpers.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_document_reference_content(attachment_data=None, url=None, content_type='text/plain', language='en-US', title=None, **kwargs)

Create a FHIR DocumentReferenceContent object.

Creates a DocumentReferenceContent structure that can be added to a DocumentReference. Either attachment_data or url must be provided. If attachment_data is provided, it will be base64 encoded automatically.

PARAMETER	DESCRIPTION
attachment_data	The content data (text that will be base64 encoded) TYPE: Optional[str] DEFAULT: None
url	URL where the content can be accessed TYPE: Optional[str] DEFAULT: None
content_type	MIME type (e.g., 'text/plain', 'text/html', 'application/pdf') (default: text/plain) TYPE: str DEFAULT: 'text/plain'
language	Language code (default: en-US) TYPE: Optional[str] DEFAULT: 'en-US'
title	Optional title for the content (default: "Attachment created by HealthChain") TYPE: Optional[str] DEFAULT: None
**kwargs	Additional DocumentReferenceContent fields (e.g., format, profile) DEFAULT: {}

RETURNS	DESCRIPTION
Dict[str, Any]	Dict[str, Any]: A FHIR DocumentReferenceContent dictionary with attachment and optional language

Example

Create content with inline data

content = create_document_reference_content( ... attachment_data="Patient presents with fever...", ... content_type="text/plain", ... title="Clinical Note" ... )

Create content with URL reference

content = create_document_reference_content( ... url="https://example.com/document.pdf", ... content_type="application/pdf", ... title="Lab Report" ... )

Add content to a DocumentReference

doc_ref = DocumentReference( ... id="doc-1", ... status="current", ... content=[content] ... )

Source code in healthchain/fhir/resourcehelpers.py

def create_document_reference_content(
    attachment_data: Optional[str] = None,
    url: Optional[str] = None,
    content_type: str = "text/plain",
    language: Optional[str] = "en-US",
    title: Optional[str] = None,
    **kwargs,
) -> Dict[str, Any]:
    """Create a FHIR DocumentReferenceContent object.

    Creates a DocumentReferenceContent structure that can be added to a DocumentReference.
    Either attachment_data or url must be provided. If attachment_data is provided, it will
    be base64 encoded automatically.

    Args:
        attachment_data: The content data (text that will be base64 encoded)
        url: URL where the content can be accessed
        content_type: MIME type (e.g., 'text/plain', 'text/html', 'application/pdf') (default: text/plain)
        language: Language code (default: en-US)
        title: Optional title for the content (default: "Attachment created by HealthChain")
        **kwargs: Additional DocumentReferenceContent fields (e.g., format, profile)

    Returns:
        Dict[str, Any]: A FHIR DocumentReferenceContent dictionary with attachment and optional language

    Example:
        >>> # Create content with inline data
        >>> content = create_document_reference_content(
        ...     attachment_data="Patient presents with fever...",
        ...     content_type="text/plain",
        ...     title="Clinical Note"
        ... )
        >>>
        >>> # Create content with URL reference
        >>> content = create_document_reference_content(
        ...     url="https://example.com/document.pdf",
        ...     content_type="application/pdf",
        ...     title="Lab Report"
        ... )
        >>>
        >>> # Add content to a DocumentReference
        >>> doc_ref = DocumentReference(
        ...     id="doc-1",
        ...     status="current",
        ...     content=[content]
        ... )
    """
    if not attachment_data and not url:
        logger.warning(
            "No attachment_data or url provided for DocumentReferenceContent"
        )

    if title is None:
        title = "Attachment created by HealthChain"

    attachment = create_single_attachment(
        content_type=content_type,
        data=attachment_data,
        url=url,
        title=title,
    )

    content: Dict[str, Any] = {
        "attachment": attachment,
    }

    if language:
        content["language"] = language

    content.update(kwargs)

    return content

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/resourcehelpers.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_patient(gender=None, birth_date=None, identifier=None, identifier_system='http://hospital.example.org')

Create a minimal FHIR Patient resource with basic gender and birthdate If you need to create a more complex patient, use the FHIR Patient resource directly https://hl7.org/fhir/patient.html (No required fields).

PARAMETER	DESCRIPTION
gender	Administrative gender (male, female, other, unknown) TYPE: Optional[str] DEFAULT: None
birth_date	Birth date in YYYY-MM-DD format TYPE: Optional[str] DEFAULT: None
identifier	Optional identifier value for the patient (e.g., MRN) TYPE: Optional[str] DEFAULT: None
identifier_system	The system for the identifier (default: "http://hospital.example.org") TYPE: Optional[str] DEFAULT: 'http://hospital.example.org'

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

Source code in healthchain/fhir/resourcehelpers.py

def create_patient(
    gender: Optional[str] = None,
    birth_date: Optional[str] = None,
    identifier: Optional[str] = None,
    identifier_system: Optional[str] = "http://hospital.example.org",
) -> Patient:
    """
    Create a minimal FHIR Patient resource with basic gender and birthdate
    If you need to create a more complex patient, use the FHIR Patient resource directly
    https://hl7.org/fhir/patient.html (No required fields).

    Args:
        gender: Administrative gender (male, female, other, unknown)
        birth_date: Birth date in YYYY-MM-DD format
        identifier: Optional identifier value for the patient (e.g., MRN)
        identifier_system: The system for the identifier (default: "http://hospital.example.org")

    Returns:
        Patient: A FHIR Patient resource with an auto-generated ID prefixed with 'hc-'
    """
    patient_id = _generate_id()

    patient_data = {"id": patient_id}

    if birth_date:
        patient_data["birthDate"] = birth_date

    if gender:
        patient_data["gender"] = gender.lower()

    if identifier:
        patient_data["identifier"] = [
            Identifier(
                system=identifier_system,
                value=identifier,
            )
        ]

    patient = Patient(**patient_data)
    return patient

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/readers.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_risk_assessment_from_prediction(subject, prediction, status='final', method=None, basis=None, comment=None, occurrence_datetime=None)

Create a FHIR RiskAssessment from ML model prediction output. If you need to create a more complex risk assessment, use the FHIR RiskAssessment resource directly. https://hl7.org/fhir/riskassessment.html

PARAMETER	DESCRIPTION
subject	REQUIRED. Reference to the patient (e.g. "Patient/123") TYPE: str
prediction	Dictionary containing prediction details with keys: - outcome: CodeableConcept or dict with code, display, system for the predicted outcome - probability: float between 0 and 1 representing the risk probability - qualitative_risk: Optional str indicating risk level (e.g., "high", "moderate", "low") TYPE: Dict[str, Any]
status	REQUIRED. The status of the assessment (default: "final") TYPE: str DEFAULT: 'final'
method	Optional CodeableConcept describing the assessment method/model used TYPE: Optional[CodeableConcept] DEFAULT: None
basis	Optional list of References to observations or other resources used as input TYPE: Optional[List[Reference]] DEFAULT: None
comment	Optional text comment about the assessment TYPE: Optional[str] DEFAULT: None
occurrence_datetime	When the assessment was made (ISO format). Uses current time if not provided. TYPE: Optional[str] DEFAULT: None

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

Example

prediction = { ... "outcome": {"code": "A41.9", "display": "Sepsis", "system": "http://hl7.org/fhir/sid/icd-10"}, ... "probability": 0.85, ... "qualitative_risk": "high" ... } risk = create_risk_assessment("Patient/123", prediction)

Source code in healthchain/fhir/resourcehelpers.py

def create_risk_assessment_from_prediction(
    subject: str,
    prediction: Dict[str, Any],
    status: str = "final",
    method: Optional[CodeableConcept] = None,
    basis: Optional[List[Reference]] = None,
    comment: Optional[str] = None,
    occurrence_datetime: Optional[str] = None,
) -> RiskAssessment:
    """
    Create a FHIR RiskAssessment from ML model prediction output.
    If you need to create a more complex risk assessment, use the FHIR RiskAssessment resource directly.
    https://hl7.org/fhir/riskassessment.html

    Args:
        subject: REQUIRED. Reference to the patient (e.g. "Patient/123")
        prediction: Dictionary containing prediction details with keys:
            - outcome: CodeableConcept or dict with code, display, system for the predicted outcome
            - probability: float between 0 and 1 representing the risk probability
            - qualitative_risk: Optional str indicating risk level (e.g., "high", "moderate", "low")
        status: REQUIRED. The status of the assessment (default: "final")
        method: Optional CodeableConcept describing the assessment method/model used
        basis: Optional list of References to observations or other resources used as input
        comment: Optional text comment about the assessment

        occurrence_datetime: When the assessment was made (ISO format). Uses current time if not provided.

    Returns:
        RiskAssessment: A FHIR RiskAssessment resource with an auto-generated ID prefixed with 'hc-'

    Example:
        >>> prediction = {
        ...     "outcome": {"code": "A41.9", "display": "Sepsis", "system": "http://hl7.org/fhir/sid/icd-10"},
        ...     "probability": 0.85,
        ...     "qualitative_risk": "high"
        ... }
        >>> risk = create_risk_assessment("Patient/123", prediction)
    """
    if not occurrence_datetime:
        occurrence_datetime = datetime.datetime.now(datetime.timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%S%z"
        )

    outcome = prediction.get("outcome")
    if isinstance(outcome, dict):
        outcome_concept = create_single_codeable_concept(
            code=outcome["code"],
            display=outcome.get("display"),
            system=outcome.get("system", "http://snomed.info/sct"),
        )
    else:
        outcome_concept = outcome

    prediction_data = {
        "outcome": outcome_concept,
    }

    if "probability" in prediction:
        prediction_data["probabilityDecimal"] = prediction["probability"]

    if "qualitative_risk" in prediction:
        prediction_data["qualitativeRisk"] = create_single_codeable_concept(
            code=prediction["qualitative_risk"],
            display=prediction["qualitative_risk"].capitalize(),
            system="http://terminology.hl7.org/CodeSystem/risk-probability",
        )

    risk_assessment_data = {
        "id": _generate_id(),
        "status": status,
        "subject": Reference(reference=subject),
        "occurrenceDateTime": occurrence_datetime,
        "prediction": [prediction_data],
    }

    if method:
        risk_assessment_data["method"] = method

    if basis:
        risk_assessment_data["basis"] = basis

    if comment:
        risk_assessment_data["note"] = [{"text": comment}]

    risk_assessment = RiskAssessment(**risk_assessment_data)

    return risk_assessment

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/elementhelpers.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/elementhelpers.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/elementhelpers.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,
        }
    ]

create_value_quantity_observation(code, value, unit, status='final', subject=None, system='http://loinc.org', display=None, effective_datetime=None)

Create a minimal FHIR Observation for vital signs or laboratory values. If you need to create a more complex observation, use the FHIR Observation resource directly. https://hl7.org/fhir/observation.html

PARAMETER	DESCRIPTION
status	REQUIRED. The status of the observation (default: "final") TYPE: str DEFAULT: 'final'
code	REQUIRED. The observation code (e.g., LOINC code for the measurement) TYPE: str
value	The numeric value of the observation TYPE: float
unit	The unit of measure (e.g., "beats/min", "mg/dL") TYPE: str
system	The code system for the observation code (default: LOINC) TYPE: str DEFAULT: 'http://loinc.org'
display	The display name for the observation code TYPE: Optional[str] DEFAULT: None
effective_datetime	When the observation was made (ISO format). Uses current time if not provided. TYPE: Optional[str] DEFAULT: None
subject	Reference to the patient (e.g. "Patient/123") TYPE: Optional[str] DEFAULT: None

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

Source code in healthchain/fhir/resourcehelpers.py

def create_value_quantity_observation(
    code: str,
    value: float,
    unit: str,
    status: str = "final",
    subject: Optional[str] = None,
    system: str = "http://loinc.org",
    display: Optional[str] = None,
    effective_datetime: Optional[str] = None,
) -> Observation:
    """
    Create a minimal FHIR Observation for vital signs or laboratory values.
    If you need to create a more complex observation, use the FHIR Observation resource directly.
    https://hl7.org/fhir/observation.html

    Args:
        status: REQUIRED. The status of the observation (default: "final")
        code: REQUIRED. The observation code (e.g., LOINC code for the measurement)
        value: The numeric value of the observation
        unit: The unit of measure (e.g., "beats/min", "mg/dL")
        system: The code system for the observation code (default: LOINC)
        display: The display name for the observation code
        effective_datetime: When the observation was made (ISO format). Uses current time if not provided.
        subject: Reference to the patient (e.g. "Patient/123")

    Returns:
        Observation: A FHIR Observation resource with an auto-generated ID prefixed with 'hc-'
    """
    if not effective_datetime:
        effective_datetime = datetime.datetime.now(datetime.timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%S%z"
        )
    subject_ref = None
    if subject is not None:
        subject_ref = Reference(reference=subject)

    observation = Observation(
        id=_generate_id(),
        status=status,
        code=create_single_codeable_concept(code, display, system),
        subject=subject_ref,
        effectiveDateTime=effective_datetime,
        valueQuantity=Quantity(
            value=value, unit=unit, system="http://unitsofmeasure.org", code=unit
        ),
    )

    return observation

encode_gender(gender)

Encode gender as integer for ML models.

Standard encoding: Male=1, Female=0, Other/Unknown=None

PARAMETER	DESCRIPTION
gender	Gender string (case-insensitive) TYPE: str

RETURNS	DESCRIPTION
Optional[int]	Encoded gender (1 for male, 0 for female, None for other/unknown)

Source code in healthchain/fhir/utilities.py

def encode_gender(gender: str) -> Optional[int]:
    """Encode gender as integer for ML models.

    Standard encoding: Male=1, Female=0, Other/Unknown=None

    Args:
        gender: Gender string (case-insensitive)

    Returns:
        Encoded gender (1 for male, 0 for female, None for other/unknown)
    """
    if not gender:
        return None

    gender_lower = gender.lower()
    if gender_lower in ["male", "m"]:
        return 1
    elif gender_lower in ["female", "f"]:
        return 0
    else:
        return None

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/bundlehelpers.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_info(resource_type)

Get detailed information about a supported resource type.

PARAMETER	DESCRIPTION
resource_type	FHIR resource type name TYPE: str

RETURNS	DESCRIPTION
Dict[str, Any]	Dictionary with resource handler information, or empty dict if unsupported

Example

info = get_resource_info("Observation") print(info["description"]) 'Clinical observations (vitals, labs)'

Source code in healthchain/fhir/dataframe.py

def get_resource_info(resource_type: str) -> Dict[str, Any]:
    """Get detailed information about a supported resource type.

    Args:
        resource_type: FHIR resource type name

    Returns:
        Dictionary with resource handler information, or empty dict if unsupported

    Example:
        >>> info = get_resource_info("Observation")
        >>> print(info["description"])
        'Clinical observations (vitals, labs)'
    """
    return SUPPORTED_RESOURCES.get(resource_type, {})

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/bundlehelpers.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)
    ]

get_supported_resources()

Get list of supported FHIR resource types.

RETURNS	DESCRIPTION
List[str]	List of resource type names that can be converted to DataFrame columns

Example

resources = get_supported_resources() print(resources) ['Patient', 'Observation', 'Condition', 'MedicationStatement']

Source code in healthchain/fhir/dataframe.py

def get_supported_resources() -> List[str]:
    """Get list of supported FHIR resource types.

    Returns:
        List of resource type names that can be converted to DataFrame columns

    Example:
        >>> resources = get_supported_resources()
        >>> print(resources)
        ['Patient', 'Observation', 'Condition', 'MedicationStatement']
    """
    return list(SUPPORTED_RESOURCES.keys())

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/bundlehelpers.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

prefetch_to_bundle(prefetch)

Flatten CDS Hooks prefetch into a collection Bundle dict.

Converts the keyed prefetch format (used in CDS Hooks) into a flat bundle suitable for Dataset.from_fhir_bundle().

PARAMETER	DESCRIPTION
prefetch	CDS Hooks prefetch dict with format: {"patient": {...}, "observations": {"entry": [...]}, ...} TYPE: Dict[str, Any]

RETURNS	DESCRIPTION
Dict[str, Any]	Bundle dict with type "collection" and flattened entries

Example

prefetch = request.prefetch bundle = prefetch_to_bundle(prefetch) dataset = Dataset.from_fhir_bundle(bundle, schema=schema)

Source code in healthchain/fhir/readers.py

def prefetch_to_bundle(prefetch: Dict[str, Any]) -> Dict[str, Any]:
    """Flatten CDS Hooks prefetch into a collection Bundle dict.

    Converts the keyed prefetch format (used in CDS Hooks) into a flat bundle
    suitable for Dataset.from_fhir_bundle().

    Args:
        prefetch: CDS Hooks prefetch dict with format:
            {"patient": {...}, "observations": {"entry": [...]}, ...}

    Returns:
        Bundle dict with type "collection" and flattened entries

    Example:
        >>> prefetch = request.prefetch
        >>> bundle = prefetch_to_bundle(prefetch)
        >>> dataset = Dataset.from_fhir_bundle(bundle, schema=schema)
    """
    entries = []
    for key, value in prefetch.items():
        if isinstance(value, dict):
            if "entry" in value:  # Searchset bundle
                entries.extend(value["entry"])
            elif "resourceType" in value:  # Single resource
                entries.append({"resource": value})
    return {"type": "collection", "entry": entries}

print_supported_resources()

Print user-friendly list of supported FHIR resources for conversion.

Example

from healthchain.fhir.converters import print_supported_resources print_supported_resources() Supported FHIR Resources for ML Dataset Conversion:

✓ Patient Patient demographics (age, gender) Columns: age, gender ...

Source code in healthchain/fhir/dataframe.py

def print_supported_resources() -> None:
    """Print user-friendly list of supported FHIR resources for conversion.

    Example:
        >>> from healthchain.fhir.converters import print_supported_resources
        >>> print_supported_resources()
        Supported FHIR Resources for ML Dataset Conversion:

          ✓ Patient
            Patient demographics (age, gender)
            Columns: age, gender
        ...
    """
    print("Supported FHIR Resources for ML Dataset Conversion:\n")
    for resource, info in SUPPORTED_RESOURCES.items():
        print(f"  ✓ {resource}")
        print(f"    {info['description']}")
        if isinstance(info["output_columns"], list):
            print(f"    Columns: {', '.join(info['output_columns'])}")
        else:
            print(f"    Columns: {info['output_columns']}")
        if info.get("options"):
            print(f"    Options: {', '.join(info['options'])}")
        print()

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/readers.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/resourcehelpers.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/bundlehelpers.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)

bundlehelpers

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/bundlehelpers.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]

count_resources(bundle)

Count resources by type in a bundle.

PARAMETER	DESCRIPTION
bundle	The FHIR Bundle to analyze TYPE: Bundle

RETURNS	DESCRIPTION
dict[str, int]	Dictionary mapping resource type names to their counts.
Example	{"Condition": 2, "MedicationStatement": 1, "Patient": 1} TYPE: dict[str, int]

Example

bundle = create_bundle() add_resource(bundle, create_condition(...)) add_resource(bundle, create_condition(...)) add_resource(bundle, create_medication_statement(...)) counts = count_resources(bundle) print(counts)

Source code in healthchain/fhir/bundlehelpers.py

def count_resources(bundle: Bundle) -> dict[str, int]:
    """Count resources by type in a bundle.

    Args:
        bundle: The FHIR Bundle to analyze

    Returns:
        Dictionary mapping resource type names to their counts.
        Example: {"Condition": 2, "MedicationStatement": 1, "Patient": 1}

    Example:
        >>> bundle = create_bundle()
        >>> add_resource(bundle, create_condition(...))
        >>> add_resource(bundle, create_condition(...))
        >>> add_resource(bundle, create_medication_statement(...))
        >>> counts = count_resources(bundle)
        >>> print(counts)
        {'Condition': 2, 'MedicationStatement': 1}
    """
    if not bundle or not bundle.entry:
        return {}

    counts: dict[str, int] = {}
    for entry in bundle.entry:
        if entry.resource:
            # Get the resource type from the class name
            resource_type = entry.resource.__resource_type__
            counts[resource_type] = counts.get(resource_type, 0) + 1

    return counts

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/bundlehelpers.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/bundlehelpers.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/bundlehelpers.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/bundlehelpers.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/bundlehelpers.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/bundlehelpers.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)

dataframe

FHIR to DataFrame converters.

This module provides generic functions to convert FHIR Bundles to pandas DataFrames for analysis and ML model deployment.

In instances where there are multiple codes present for a single resource, the first code is used as the primary code.

BundleConverterConfig

Bases: BaseModel

Configuration for FHIR Bundle to DataFrame conversion.

This configuration object controls which FHIR resources are processed and how they are converted to DataFrame columns for ML model deployment.

ATTRIBUTE	DESCRIPTION
resources	List of FHIR resource types to include in the conversion TYPE: List[str]
observation_aggregation	How to aggregate multiple observation values TYPE: Literal['mean', 'median', 'max', 'min', 'last']
age_calculation	Method for calculating patient age TYPE: Literal['current_date', 'event_date']
event_date_source	Which resource to extract event date from TYPE: Literal['Observation', 'Encounter']
event_date_strategy	Which date to use when multiple dates exist TYPE: Literal['earliest', 'latest', 'first']
resource_options	Resource-specific configuration options (extensible) TYPE: Dict[str, Dict[str, Any]]

Example

config = BundleConverterConfig( ... resources=["Patient", "Observation", "Condition"], ... observation_aggregation="median" ... ) df = bundle_to_dataframe(bundle, config=config)

Source code in healthchain/fhir/dataframe.py

class BundleConverterConfig(BaseModel):
    """Configuration for FHIR Bundle to DataFrame conversion.

    This configuration object controls which FHIR resources are processed and how
    they are converted to DataFrame columns for ML model deployment.

    Attributes:
        resources: List of FHIR resource types to include in the conversion
        observation_aggregation: How to aggregate multiple observation values
        age_calculation: Method for calculating patient age
        event_date_source: Which resource to extract event date from
        event_date_strategy: Which date to use when multiple dates exist
        resource_options: Resource-specific configuration options (extensible)

    Example:
        >>> config = BundleConverterConfig(
        ...     resources=["Patient", "Observation", "Condition"],
        ...     observation_aggregation="median"
        ... )
        >>> df = bundle_to_dataframe(bundle, config=config)
    """

    # Core resources to include
    resources: List[str] = ["Patient", "Observation"]

    # Observation-specific options
    observation_aggregation: Literal["mean", "median", "max", "min", "last"] = "mean"

    # Patient age calculation
    age_calculation: Literal["current_date", "event_date"] = "current_date"
    event_date_source: Literal["Observation", "Encounter"] = "Observation"
    event_date_strategy: Literal["earliest", "latest", "first"] = "earliest"

    # Resource-specific options (extensible for future use)
    resource_options: Dict[str, Dict[str, Any]] = {}

    model_config = ConfigDict(extra="allow")

    @field_validator("resources")
    @classmethod
    def validate_resources(cls, v):
        """Validate that requested resources are supported and warn about unsupported ones."""
        supported = get_supported_resources()
        unsupported = [r for r in v if r not in supported]
        if unsupported:
            logger.warning(
                f"Unsupported resources will be skipped: {unsupported}. "
                f"Supported resources: {supported}"
            )
        return v

validate_resources(v) classmethod

Validate that requested resources are supported and warn about unsupported ones.

Source code in healthchain/fhir/dataframe.py

@field_validator("resources")
@classmethod
def validate_resources(cls, v):
    """Validate that requested resources are supported and warn about unsupported ones."""
    supported = get_supported_resources()
    unsupported = [r for r in v if r not in supported]
    if unsupported:
        logger.warning(
            f"Unsupported resources will be skipped: {unsupported}. "
            f"Supported resources: {supported}"
        )
    return v

bundle_to_dataframe(bundle, config=None)

Convert a FHIR Bundle to a pandas DataFrame.

Converts FHIR resources to a tabular format with one row per patient. Uses a configuration object to control which resources are processed and how.

PARAMETER	DESCRIPTION
bundle	FHIR Bundle resource (object or dict) TYPE: Union[Bundle, Dict[str, Any]]
config	BundleConverterConfig object specifying conversion behavior. If None, uses default config (Patient + Observation with mean aggregation) TYPE: Optional[BundleConverterConfig] DEFAULT: None

RETURNS	DESCRIPTION
DataFrame	DataFrame with one row per patient and columns for each feature

Example

from healthchain.fhir.converters import BundleConverterConfig

Default behavior

df = bundle_to_dataframe(bundle)

Custom config

config = BundleConverterConfig( ... resources=["Patient", "Observation", "Condition"], ... observation_aggregation="median", ... age_calculation="event_date" ... ) df = bundle_to_dataframe(bundle, config=config)

Source code in healthchain/fhir/dataframe.py

def bundle_to_dataframe(
    bundle: Union[Bundle, Dict[str, Any]],
    config: Optional[BundleConverterConfig] = None,
) -> pd.DataFrame:
    """Convert a FHIR Bundle to a pandas DataFrame.

    Converts FHIR resources to a tabular format with one row per patient.
    Uses a configuration object to control which resources are processed and how.

    Args:
        bundle: FHIR Bundle resource (object or dict)
        config: BundleConverterConfig object specifying conversion behavior.
            If None, uses default config (Patient + Observation with mean aggregation)

    Returns:
        DataFrame with one row per patient and columns for each feature

    Example:
        >>> from healthchain.fhir.converters import BundleConverterConfig
        >>>
        >>> # Default behavior
        >>> df = bundle_to_dataframe(bundle)
        >>>
        >>> # Custom config
        >>> config = BundleConverterConfig(
        ...     resources=["Patient", "Observation", "Condition"],
        ...     observation_aggregation="median",
        ...     age_calculation="event_date"
        ... )
        >>> df = bundle_to_dataframe(bundle, config=config)
    """
    # Use default config if not provided
    if config is None:
        config = BundleConverterConfig()

    # Group resources by patient
    patient_data = group_bundle_by_patient(bundle)

    if not patient_data:
        return pd.DataFrame()

    # Build rows for each patient
    rows = []
    for patient_ref, resources in patient_data.items():
        row = {"patient_ref": patient_ref}

        # Process each requested resource type using registry
        for resource_type in config.resources:
            handler_info = SUPPORTED_RESOURCES.get(resource_type)

            if not handler_info:
                # Skip unsupported resources gracefully (already warned by validator)
                continue

            # Get handler function by name
            handler_name = handler_info["handler"]
            handler = globals()[handler_name]

            # Call handler with standardized signature
            features = handler(resources, config)
            if features:
                row.update(features)

        rows.append(row)

    return pd.DataFrame(rows)

extract_event_date(resources, source='Observation', strategy='earliest')

Extract event date from patient resources for age calculation.

Used primarily for MIMIC-IV on FHIR datasets where age is calculated based on event dates rather than current date.

PARAMETER	DESCRIPTION
resources	Dictionary of patient resources (from group_bundle_by_patient) TYPE: Dict[str, List[Any]]
source	Which resource type to extract date from ("Observation" or "Encounter") TYPE: str DEFAULT: 'Observation'
strategy	Which date to use ("earliest", "latest", "first") TYPE: str DEFAULT: 'earliest'

RETURNS	DESCRIPTION
Optional[str]	Event date in ISO format, or None if no suitable date found

Example

resources = {"observations": [obs1, obs2], "encounters": [enc1]} event_date = extract_event_date(resources, source="Observation", strategy="earliest")

Source code in healthchain/fhir/dataframe.py

def extract_event_date(
    resources: Dict[str, List[Any]],
    source: str = "Observation",
    strategy: str = "earliest",
) -> Optional[str]:
    """Extract event date from patient resources for age calculation.

    Used primarily for MIMIC-IV on FHIR datasets where age is calculated
    based on event dates rather than current date.

    Args:
        resources: Dictionary of patient resources (from group_bundle_by_patient)
        source: Which resource type to extract date from ("Observation" or "Encounter")
        strategy: Which date to use ("earliest", "latest", "first")

    Returns:
        Event date in ISO format, or None if no suitable date found

    Example:
        >>> resources = {"observations": [obs1, obs2], "encounters": [enc1]}
        >>> event_date = extract_event_date(resources, source="Observation", strategy="earliest")
    """
    if source == "Observation":
        items = resources.get("observations", [])
        date_field = "effectiveDateTime"
    elif source == "Encounter":
        items = resources.get("encounters", [])
        date_field = "period"
    else:
        return None

    if not items:
        return None

    dates = []
    for item in items:
        if source == "Encounter":
            # Extract start date from period
            period = _get_field(item, date_field)
            if period:
                start = _get_field(period, "start")
                if start:
                    dates.append(start)
        else:
            # Direct date field
            date_value = _get_field(item, date_field)
            if date_value:
                dates.append(date_value)

    if not dates:
        return None

    # Apply strategy
    if strategy == "earliest":
        return min(dates)
    elif strategy == "latest":
        return max(dates)
    elif strategy == "first":
        return dates[0]
    else:
        return min(dates)  # Default to earliest

extract_observation_value(observation)

Extract numeric value from an Observation dict.

Handles different value types (valueQuantity, valueInteger, valueString) and attempts to convert to float.

Source code in healthchain/fhir/dataframe.py

def extract_observation_value(observation: Dict) -> Optional[float]:
    """Extract numeric value from an Observation dict.

    Handles different value types (valueQuantity, valueInteger, valueString) and
    attempts to convert to float.
    """

    try:
        value_quantity = _get_field(observation, "valueQuantity")
        if value_quantity:
            value = _get_field(value_quantity, "value")
            if value is not None:
                return float(value)

        value_int = _get_field(observation, "valueInteger")
        if value_int is not None:
            return float(value_int)

        value_str = _get_field(observation, "valueString")
        if value_str:
            return float(value_str)

    except (ValueError, TypeError):
        pass

    return None

get_resource_info(resource_type)

Get detailed information about a supported resource type.

PARAMETER	DESCRIPTION
resource_type	FHIR resource type name TYPE: str

RETURNS	DESCRIPTION
Dict[str, Any]	Dictionary with resource handler information, or empty dict if unsupported

Example

info = get_resource_info("Observation") print(info["description"]) 'Clinical observations (vitals, labs)'

Source code in healthchain/fhir/dataframe.py

def get_resource_info(resource_type: str) -> Dict[str, Any]:
    """Get detailed information about a supported resource type.

    Args:
        resource_type: FHIR resource type name

    Returns:
        Dictionary with resource handler information, or empty dict if unsupported

    Example:
        >>> info = get_resource_info("Observation")
        >>> print(info["description"])
        'Clinical observations (vitals, labs)'
    """
    return SUPPORTED_RESOURCES.get(resource_type, {})

get_supported_resources()

Get list of supported FHIR resource types.

RETURNS	DESCRIPTION
List[str]	List of resource type names that can be converted to DataFrame columns

Example

resources = get_supported_resources() print(resources) ['Patient', 'Observation', 'Condition', 'MedicationStatement']

Source code in healthchain/fhir/dataframe.py

def get_supported_resources() -> List[str]:
    """Get list of supported FHIR resource types.

    Returns:
        List of resource type names that can be converted to DataFrame columns

    Example:
        >>> resources = get_supported_resources()
        >>> print(resources)
        ['Patient', 'Observation', 'Condition', 'MedicationStatement']
    """
    return list(SUPPORTED_RESOURCES.keys())

group_bundle_by_patient(bundle)

Group Bundle resources by patient reference.

Organizes FHIR resources in a Bundle by their associated patient, making it easier to process patient-centric data. Accepts both Pydantic Bundle objects and dicts, converts to dict internally for performance.

PARAMETER	DESCRIPTION
bundle	FHIR Bundle resource (Pydantic object or dict) TYPE: Union[Bundle, Dict[str, Any]]

RETURNS	DESCRIPTION
Dict[str, Dict[str, List[Any]]]	Dictionary mapping patient references to their resources:
Dict[str, Dict[str, List[Any]]]	{ "Patient/123": { "patient": Patient resource dict, "observations": [Observation dict, ...], "conditions": [Condition dict, ...], ... }
Dict[str, Dict[str, List[Any]]]	}

Source code in healthchain/fhir/dataframe.py

def group_bundle_by_patient(
    bundle: Union[Bundle, Dict[str, Any]],
) -> Dict[str, Dict[str, List[Any]]]:
    """Group Bundle resources by patient reference.

    Organizes FHIR resources in a Bundle by their associated patient, making it easier
    to process patient-centric data. Accepts both Pydantic Bundle objects and dicts,
    converts to dict internally for performance.

    Args:
        bundle: FHIR Bundle resource (Pydantic object or dict)

    Returns:
        Dictionary mapping patient references to their resources:
        {
            "Patient/123": {
                "patient": Patient resource dict,
                "observations": [Observation dict, ...],
                "conditions": [Condition dict, ...],
                ...
            }
        }
    """
    if not isinstance(bundle, dict):
        bundle = bundle.model_dump()

    patient_data = defaultdict(
        lambda: {
            "patient": None,
            "observations": [],
            "conditions": [],
            "medications": [],
            "allergies": [],
            "procedures": [],
            "encounters": [],
            "other": [],
        }
    )

    # Get bundle entries
    entries = _get_field(bundle, "entry")
    if not entries:
        return dict(patient_data)

    for entry in entries:
        # Get resource from entry
        resource = _get_field(entry, "resource")
        if not resource:
            continue

        resource_type = _get_field(resource, "resourceType")
        resource_id = _get_field(resource, "id")

        if resource_type == "Patient":
            patient_ref = f"Patient/{resource_id}"
            patient_data[patient_ref]["patient"] = resource

        else:
            # Get patient reference from resource
            subject = _get_field(resource, "subject")
            patient_field = _get_field(resource, "patient")

            patient_ref = _get_reference(subject) or _get_reference(patient_field)

            if patient_ref:
                # Add to appropriate list based on resource type
                if resource_type == "Observation":
                    patient_data[patient_ref]["observations"].append(resource)
                elif resource_type == "Condition":
                    patient_data[patient_ref]["conditions"].append(resource)
                elif resource_type == "MedicationStatement":
                    patient_data[patient_ref]["medications"].append(resource)
                elif resource_type == "AllergyIntolerance":
                    patient_data[patient_ref]["allergies"].append(resource)
                elif resource_type == "Procedure":
                    patient_data[patient_ref]["procedures"].append(resource)
                elif resource_type == "Encounter":
                    patient_data[patient_ref]["encounters"].append(resource)
                else:
                    patient_data[patient_ref]["other"].append(resource)

    return dict(patient_data)

print_supported_resources()

Print user-friendly list of supported FHIR resources for conversion.

Example

from healthchain.fhir.converters import print_supported_resources print_supported_resources() Supported FHIR Resources for ML Dataset Conversion:

✓ Patient Patient demographics (age, gender) Columns: age, gender ...

Source code in healthchain/fhir/dataframe.py

def print_supported_resources() -> None:
    """Print user-friendly list of supported FHIR resources for conversion.

    Example:
        >>> from healthchain.fhir.converters import print_supported_resources
        >>> print_supported_resources()
        Supported FHIR Resources for ML Dataset Conversion:

          ✓ Patient
            Patient demographics (age, gender)
            Columns: age, gender
        ...
    """
    print("Supported FHIR Resources for ML Dataset Conversion:\n")
    for resource, info in SUPPORTED_RESOURCES.items():
        print(f"  ✓ {resource}")
        print(f"    {info['description']}")
        if isinstance(info["output_columns"], list):
            print(f"    Columns: {', '.join(info['output_columns'])}")
        else:
            print(f"    Columns: {info['output_columns']}")
        if info.get("options"):
            print(f"    Options: {', '.join(info['options'])}")
        print()

elementhelpers

FHIR element creation functions.

This module provides convenience functions for creating FHIR elements that are used as building blocks within FHIR resources (e.g., CodeableConcept, Attachment, Coding).

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/elementhelpers.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/elementhelpers.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/elementhelpers.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,
        }
    ]

readers

FHIR conversion and reading functions.

This module provides functions for converting between different FHIR representations and reading data from FHIR resources.

convert_prefetch_to_fhir_objects(prefetch_dict)

Convert a dictionary of FHIR resource dicts to FHIR Resource objects.

Takes a prefetch dictionary where values may be either dict representations of FHIR resources or already instantiated FHIR Resource objects, and ensures all values are FHIR Resource objects.

PARAMETER	DESCRIPTION
prefetch_dict	Dictionary mapping keys to FHIR resource dicts or objects TYPE: Dict[str, Any]

RETURNS	DESCRIPTION
Dict[str, Resource]	Dict[str, Resource]: Dictionary with same keys but all values as FHIR Resource objects

Example

prefetch = { ... "patient": {"resourceType": "Patient", "id": "123"}, ... "condition": Condition(id="456", ...) ... } fhir_objects = convert_prefetch_to_fhir_objects(prefetch) isinstance(fhir_objects["patient"], Patient) # True isinstance(fhir_objects["condition"], Condition) # True

Source code in healthchain/fhir/readers.py

def convert_prefetch_to_fhir_objects(
    prefetch_dict: Dict[str, Any],
) -> Dict[str, Resource]:
    """Convert a dictionary of FHIR resource dicts to FHIR Resource objects.

    Takes a prefetch dictionary where values may be either dict representations of FHIR
    resources or already instantiated FHIR Resource objects, and ensures all values are
    FHIR Resource objects.

    Args:
        prefetch_dict: Dictionary mapping keys to FHIR resource dicts or objects

    Returns:
        Dict[str, Resource]: Dictionary with same keys but all values as FHIR Resource objects

    Example:
        >>> prefetch = {
        ...     "patient": {"resourceType": "Patient", "id": "123"},
        ...     "condition": Condition(id="456", ...)
        ... }
        >>> fhir_objects = convert_prefetch_to_fhir_objects(prefetch)
        >>> isinstance(fhir_objects["patient"], Patient)  # True
        >>> isinstance(fhir_objects["condition"], Condition)  # True
    """
    from fhir.resources import get_fhir_model_class

    result: Dict[str, Resource] = {}

    for key, resource_data in prefetch_dict.items():
        if isinstance(resource_data, dict):
            # Convert dict to FHIR Resource object
            resource_type = resource_data.get("resourceType")
            if resource_type:
                try:
                    resource_class = get_fhir_model_class(resource_type)
                    result[key] = resource_class(**resource_data)
                except Exception as e:
                    logger.warning(
                        f"Failed to convert {resource_type} to FHIR object: {e}"
                    )
                    result[key] = resource_data
            else:
                logger.warning(
                    f"No resourceType found for key '{key}', keeping as dict"
                )
                result[key] = resource_data
        elif isinstance(resource_data, Resource):
            # Already a FHIR object
            result[key] = resource_data
        else:
            logger.warning(f"Unexpected type for key '{key}': {type(resource_data)}")
            result[key] = resource_data

    return result

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/readers.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

prefetch_to_bundle(prefetch)

Flatten CDS Hooks prefetch into a collection Bundle dict.

Converts the keyed prefetch format (used in CDS Hooks) into a flat bundle suitable for Dataset.from_fhir_bundle().

PARAMETER	DESCRIPTION
prefetch	CDS Hooks prefetch dict with format: {"patient": {...}, "observations": {"entry": [...]}, ...} TYPE: Dict[str, Any]

RETURNS	DESCRIPTION
Dict[str, Any]	Bundle dict with type "collection" and flattened entries

Example

prefetch = request.prefetch bundle = prefetch_to_bundle(prefetch) dataset = Dataset.from_fhir_bundle(bundle, schema=schema)

Source code in healthchain/fhir/readers.py

def prefetch_to_bundle(prefetch: Dict[str, Any]) -> Dict[str, Any]:
    """Flatten CDS Hooks prefetch into a collection Bundle dict.

    Converts the keyed prefetch format (used in CDS Hooks) into a flat bundle
    suitable for Dataset.from_fhir_bundle().

    Args:
        prefetch: CDS Hooks prefetch dict with format:
            {"patient": {...}, "observations": {"entry": [...]}, ...}

    Returns:
        Bundle dict with type "collection" and flattened entries

    Example:
        >>> prefetch = request.prefetch
        >>> bundle = prefetch_to_bundle(prefetch)
        >>> dataset = Dataset.from_fhir_bundle(bundle, schema=schema)
    """
    entries = []
    for key, value in prefetch.items():
        if isinstance(value, dict):
            if "entry" in value:  # Searchset bundle
                entries.extend(value["entry"])
            elif "resourceType" in value:  # Single resource
                entries.append({"resource": value})
    return {"type": "collection", "entry": entries}

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/readers.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

resourcehelpers

FHIR resource creation and modification functions.

This module provides convenience functions for creating and modifying FHIR resources.

Patterns: - create_(): create a new FHIR resource with sensible defaults - set_(): set specific fields of resources with soft validation - add_*(): add data to resources safely

Parameters marked REQUIRED are required by FHIR specification.

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/resourcehelpers.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/resourcehelpers.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/resourcehelpers.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/resourcehelpers.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/resourcehelpers.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_document_reference_content(attachment_data=None, url=None, content_type='text/plain', language='en-US', title=None, **kwargs)

Create a FHIR DocumentReferenceContent object.

Creates a DocumentReferenceContent structure that can be added to a DocumentReference. Either attachment_data or url must be provided. If attachment_data is provided, it will be base64 encoded automatically.

PARAMETER	DESCRIPTION
attachment_data	The content data (text that will be base64 encoded) TYPE: Optional[str] DEFAULT: None
url	URL where the content can be accessed TYPE: Optional[str] DEFAULT: None
content_type	MIME type (e.g., 'text/plain', 'text/html', 'application/pdf') (default: text/plain) TYPE: str DEFAULT: 'text/plain'
language	Language code (default: en-US) TYPE: Optional[str] DEFAULT: 'en-US'
title	Optional title for the content (default: "Attachment created by HealthChain") TYPE: Optional[str] DEFAULT: None
**kwargs	Additional DocumentReferenceContent fields (e.g., format, profile) DEFAULT: {}

RETURNS	DESCRIPTION
Dict[str, Any]	Dict[str, Any]: A FHIR DocumentReferenceContent dictionary with attachment and optional language

Example

Create content with inline data

content = create_document_reference_content( ... attachment_data="Patient presents with fever...", ... content_type="text/plain", ... title="Clinical Note" ... )

Create content with URL reference

content = create_document_reference_content( ... url="https://example.com/document.pdf", ... content_type="application/pdf", ... title="Lab Report" ... )

Add content to a DocumentReference

doc_ref = DocumentReference( ... id="doc-1", ... status="current", ... content=[content] ... )

Source code in healthchain/fhir/resourcehelpers.py

def create_document_reference_content(
    attachment_data: Optional[str] = None,
    url: Optional[str] = None,
    content_type: str = "text/plain",
    language: Optional[str] = "en-US",
    title: Optional[str] = None,
    **kwargs,
) -> Dict[str, Any]:
    """Create a FHIR DocumentReferenceContent object.

    Creates a DocumentReferenceContent structure that can be added to a DocumentReference.
    Either attachment_data or url must be provided. If attachment_data is provided, it will
    be base64 encoded automatically.

    Args:
        attachment_data: The content data (text that will be base64 encoded)
        url: URL where the content can be accessed
        content_type: MIME type (e.g., 'text/plain', 'text/html', 'application/pdf') (default: text/plain)
        language: Language code (default: en-US)
        title: Optional title for the content (default: "Attachment created by HealthChain")
        **kwargs: Additional DocumentReferenceContent fields (e.g., format, profile)

    Returns:
        Dict[str, Any]: A FHIR DocumentReferenceContent dictionary with attachment and optional language

    Example:
        >>> # Create content with inline data
        >>> content = create_document_reference_content(
        ...     attachment_data="Patient presents with fever...",
        ...     content_type="text/plain",
        ...     title="Clinical Note"
        ... )
        >>>
        >>> # Create content with URL reference
        >>> content = create_document_reference_content(
        ...     url="https://example.com/document.pdf",
        ...     content_type="application/pdf",
        ...     title="Lab Report"
        ... )
        >>>
        >>> # Add content to a DocumentReference
        >>> doc_ref = DocumentReference(
        ...     id="doc-1",
        ...     status="current",
        ...     content=[content]
        ... )
    """
    if not attachment_data and not url:
        logger.warning(
            "No attachment_data or url provided for DocumentReferenceContent"
        )

    if title is None:
        title = "Attachment created by HealthChain"

    attachment = create_single_attachment(
        content_type=content_type,
        data=attachment_data,
        url=url,
        title=title,
    )

    content: Dict[str, Any] = {
        "attachment": attachment,
    }

    if language:
        content["language"] = language

    content.update(kwargs)

    return content

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/resourcehelpers.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_patient(gender=None, birth_date=None, identifier=None, identifier_system='http://hospital.example.org')

Create a minimal FHIR Patient resource with basic gender and birthdate If you need to create a more complex patient, use the FHIR Patient resource directly https://hl7.org/fhir/patient.html (No required fields).

PARAMETER	DESCRIPTION
gender	Administrative gender (male, female, other, unknown) TYPE: Optional[str] DEFAULT: None
birth_date	Birth date in YYYY-MM-DD format TYPE: Optional[str] DEFAULT: None
identifier	Optional identifier value for the patient (e.g., MRN) TYPE: Optional[str] DEFAULT: None
identifier_system	The system for the identifier (default: "http://hospital.example.org") TYPE: Optional[str] DEFAULT: 'http://hospital.example.org'

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

Source code in healthchain/fhir/resourcehelpers.py

def create_patient(
    gender: Optional[str] = None,
    birth_date: Optional[str] = None,
    identifier: Optional[str] = None,
    identifier_system: Optional[str] = "http://hospital.example.org",
) -> Patient:
    """
    Create a minimal FHIR Patient resource with basic gender and birthdate
    If you need to create a more complex patient, use the FHIR Patient resource directly
    https://hl7.org/fhir/patient.html (No required fields).

    Args:
        gender: Administrative gender (male, female, other, unknown)
        birth_date: Birth date in YYYY-MM-DD format
        identifier: Optional identifier value for the patient (e.g., MRN)
        identifier_system: The system for the identifier (default: "http://hospital.example.org")

    Returns:
        Patient: A FHIR Patient resource with an auto-generated ID prefixed with 'hc-'
    """
    patient_id = _generate_id()

    patient_data = {"id": patient_id}

    if birth_date:
        patient_data["birthDate"] = birth_date

    if gender:
        patient_data["gender"] = gender.lower()

    if identifier:
        patient_data["identifier"] = [
            Identifier(
                system=identifier_system,
                value=identifier,
            )
        ]

    patient = Patient(**patient_data)
    return patient

create_risk_assessment_from_prediction(subject, prediction, status='final', method=None, basis=None, comment=None, occurrence_datetime=None)

Create a FHIR RiskAssessment from ML model prediction output. If you need to create a more complex risk assessment, use the FHIR RiskAssessment resource directly. https://hl7.org/fhir/riskassessment.html

PARAMETER	DESCRIPTION
subject	REQUIRED. Reference to the patient (e.g. "Patient/123") TYPE: str
prediction	Dictionary containing prediction details with keys: - outcome: CodeableConcept or dict with code, display, system for the predicted outcome - probability: float between 0 and 1 representing the risk probability - qualitative_risk: Optional str indicating risk level (e.g., "high", "moderate", "low") TYPE: Dict[str, Any]
status	REQUIRED. The status of the assessment (default: "final") TYPE: str DEFAULT: 'final'
method	Optional CodeableConcept describing the assessment method/model used TYPE: Optional[CodeableConcept] DEFAULT: None
basis	Optional list of References to observations or other resources used as input TYPE: Optional[List[Reference]] DEFAULT: None
comment	Optional text comment about the assessment TYPE: Optional[str] DEFAULT: None
occurrence_datetime	When the assessment was made (ISO format). Uses current time if not provided. TYPE: Optional[str] DEFAULT: None

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

Example

prediction = { ... "outcome": {"code": "A41.9", "display": "Sepsis", "system": "http://hl7.org/fhir/sid/icd-10"}, ... "probability": 0.85, ... "qualitative_risk": "high" ... } risk = create_risk_assessment("Patient/123", prediction)

Source code in healthchain/fhir/resourcehelpers.py

def create_risk_assessment_from_prediction(
    subject: str,
    prediction: Dict[str, Any],
    status: str = "final",
    method: Optional[CodeableConcept] = None,
    basis: Optional[List[Reference]] = None,
    comment: Optional[str] = None,
    occurrence_datetime: Optional[str] = None,
) -> RiskAssessment:
    """
    Create a FHIR RiskAssessment from ML model prediction output.
    If you need to create a more complex risk assessment, use the FHIR RiskAssessment resource directly.
    https://hl7.org/fhir/riskassessment.html

    Args:
        subject: REQUIRED. Reference to the patient (e.g. "Patient/123")
        prediction: Dictionary containing prediction details with keys:
            - outcome: CodeableConcept or dict with code, display, system for the predicted outcome
            - probability: float between 0 and 1 representing the risk probability
            - qualitative_risk: Optional str indicating risk level (e.g., "high", "moderate", "low")
        status: REQUIRED. The status of the assessment (default: "final")
        method: Optional CodeableConcept describing the assessment method/model used
        basis: Optional list of References to observations or other resources used as input
        comment: Optional text comment about the assessment

        occurrence_datetime: When the assessment was made (ISO format). Uses current time if not provided.

    Returns:
        RiskAssessment: A FHIR RiskAssessment resource with an auto-generated ID prefixed with 'hc-'

    Example:
        >>> prediction = {
        ...     "outcome": {"code": "A41.9", "display": "Sepsis", "system": "http://hl7.org/fhir/sid/icd-10"},
        ...     "probability": 0.85,
        ...     "qualitative_risk": "high"
        ... }
        >>> risk = create_risk_assessment("Patient/123", prediction)
    """
    if not occurrence_datetime:
        occurrence_datetime = datetime.datetime.now(datetime.timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%S%z"
        )

    outcome = prediction.get("outcome")
    if isinstance(outcome, dict):
        outcome_concept = create_single_codeable_concept(
            code=outcome["code"],
            display=outcome.get("display"),
            system=outcome.get("system", "http://snomed.info/sct"),
        )
    else:
        outcome_concept = outcome

    prediction_data = {
        "outcome": outcome_concept,
    }

    if "probability" in prediction:
        prediction_data["probabilityDecimal"] = prediction["probability"]

    if "qualitative_risk" in prediction:
        prediction_data["qualitativeRisk"] = create_single_codeable_concept(
            code=prediction["qualitative_risk"],
            display=prediction["qualitative_risk"].capitalize(),
            system="http://terminology.hl7.org/CodeSystem/risk-probability",
        )

    risk_assessment_data = {
        "id": _generate_id(),
        "status": status,
        "subject": Reference(reference=subject),
        "occurrenceDateTime": occurrence_datetime,
        "prediction": [prediction_data],
    }

    if method:
        risk_assessment_data["method"] = method

    if basis:
        risk_assessment_data["basis"] = basis

    if comment:
        risk_assessment_data["note"] = [{"text": comment}]

    risk_assessment = RiskAssessment(**risk_assessment_data)

    return risk_assessment

create_value_quantity_observation(code, value, unit, status='final', subject=None, system='http://loinc.org', display=None, effective_datetime=None)

Create a minimal FHIR Observation for vital signs or laboratory values. If you need to create a more complex observation, use the FHIR Observation resource directly. https://hl7.org/fhir/observation.html

PARAMETER	DESCRIPTION
status	REQUIRED. The status of the observation (default: "final") TYPE: str DEFAULT: 'final'
code	REQUIRED. The observation code (e.g., LOINC code for the measurement) TYPE: str
value	The numeric value of the observation TYPE: float
unit	The unit of measure (e.g., "beats/min", "mg/dL") TYPE: str
system	The code system for the observation code (default: LOINC) TYPE: str DEFAULT: 'http://loinc.org'
display	The display name for the observation code TYPE: Optional[str] DEFAULT: None
effective_datetime	When the observation was made (ISO format). Uses current time if not provided. TYPE: Optional[str] DEFAULT: None
subject	Reference to the patient (e.g. "Patient/123") TYPE: Optional[str] DEFAULT: None

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

Source code in healthchain/fhir/resourcehelpers.py

def create_value_quantity_observation(
    code: str,
    value: float,
    unit: str,
    status: str = "final",
    subject: Optional[str] = None,
    system: str = "http://loinc.org",
    display: Optional[str] = None,
    effective_datetime: Optional[str] = None,
) -> Observation:
    """
    Create a minimal FHIR Observation for vital signs or laboratory values.
    If you need to create a more complex observation, use the FHIR Observation resource directly.
    https://hl7.org/fhir/observation.html

    Args:
        status: REQUIRED. The status of the observation (default: "final")
        code: REQUIRED. The observation code (e.g., LOINC code for the measurement)
        value: The numeric value of the observation
        unit: The unit of measure (e.g., "beats/min", "mg/dL")
        system: The code system for the observation code (default: LOINC)
        display: The display name for the observation code
        effective_datetime: When the observation was made (ISO format). Uses current time if not provided.
        subject: Reference to the patient (e.g. "Patient/123")

    Returns:
        Observation: A FHIR Observation resource with an auto-generated ID prefixed with 'hc-'
    """
    if not effective_datetime:
        effective_datetime = datetime.datetime.now(datetime.timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%S%z"
        )
    subject_ref = None
    if subject is not None:
        subject_ref = Reference(reference=subject)

    observation = Observation(
        id=_generate_id(),
        status=status,
        code=create_single_codeable_concept(code, display, system),
        subject=subject_ref,
        effectiveDateTime=effective_datetime,
        valueQuantity=Quantity(
            value=value, unit=unit, system="http://unitsofmeasure.org", code=unit
        ),
    )

    return observation

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/resourcehelpers.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

utilities

FHIR utility functions.

This module provides utility functions for common operations like ID generation, age calculation, and gender encoding.

calculate_age_from_birthdate(birth_date)

Calculate age in years from a birth date string.

PARAMETER	DESCRIPTION
birth_date	Birth date in ISO format (YYYY-MM-DD or full ISO datetime) TYPE: str

RETURNS	DESCRIPTION
Optional[int]	Age in years, or None if birth date is invalid

Source code in healthchain/fhir/utilities.py

def calculate_age_from_birthdate(birth_date: str) -> Optional[int]:
    """Calculate age in years from a birth date string.

    Args:
        birth_date: Birth date in ISO format (YYYY-MM-DD or full ISO datetime)

    Returns:
        Age in years, or None if birth date is invalid
    """
    if not birth_date:
        return None

    try:
        if isinstance(birth_date, str):
            # Remove timezone info for simpler parsing
            birth_date_clean = birth_date.replace("Z", "").split("T")[0]
            birth_dt = datetime.datetime.strptime(birth_date_clean, "%Y-%m-%d")
        else:
            birth_dt = birth_date

        # Calculate age
        today = datetime.datetime.now()
        age = today.year - birth_dt.year

        # Adjust if birthday hasn't occurred this year
        if (today.month, today.day) < (birth_dt.month, birth_dt.day):
            age -= 1

        return age
    except (ValueError, AttributeError, TypeError):
        return None

calculate_age_from_event_date(birth_date, event_date)

Calculate age in years from birth date and event date (MIMIC-IV style).

Uses the formula: age = year(eventDate) - year(birthDate) This matches MIMIC-IV on FHIR de-identified age calculation.

PARAMETER	DESCRIPTION
birth_date	Birth date in ISO format (YYYY-MM-DD or full ISO datetime) TYPE: str
event_date	Event date in ISO format (YYYY-MM-DD or full ISO datetime) TYPE: str

RETURNS	DESCRIPTION
Optional[int]	Age in years based on year difference, or None if dates are invalid

Example

calculate_age_from_event_date("1990-06-15", "2020-03-10") 30

Source code in healthchain/fhir/utilities.py

def calculate_age_from_event_date(birth_date: str, event_date: str) -> Optional[int]:
    """Calculate age in years from birth date and event date (MIMIC-IV style).

    Uses the formula: age = year(eventDate) - year(birthDate)
    This matches MIMIC-IV on FHIR de-identified age calculation.

    Args:
        birth_date: Birth date in ISO format (YYYY-MM-DD or full ISO datetime)
        event_date: Event date in ISO format (YYYY-MM-DD or full ISO datetime)

    Returns:
        Age in years based on year difference, or None if dates are invalid

    Example:
        >>> calculate_age_from_event_date("1990-06-15", "2020-03-10")
        30
    """
    if not birth_date or not event_date:
        return None

    try:
        # Parse birth date
        if isinstance(birth_date, str):
            birth_date_clean = birth_date.replace("Z", "").split("T")[0]
            birth_year = int(birth_date_clean.split("-")[0])
        else:
            birth_year = birth_date.year

        # Parse event date
        if isinstance(event_date, str):
            event_date_clean = event_date.replace("Z", "").split("T")[0]
            event_year = int(event_date_clean.split("-")[0])
        else:
            event_year = event_date.year

        # MIMIC-IV style: simple year difference
        age = event_year - birth_year

        return age
    except (ValueError, AttributeError, TypeError, IndexError):
        return None

encode_gender(gender)

Encode gender as integer for ML models.

Standard encoding: Male=1, Female=0, Other/Unknown=None

PARAMETER	DESCRIPTION
gender	Gender string (case-insensitive) TYPE: str

RETURNS	DESCRIPTION
Optional[int]	Encoded gender (1 for male, 0 for female, None for other/unknown)

Source code in healthchain/fhir/utilities.py

def encode_gender(gender: str) -> Optional[int]:
    """Encode gender as integer for ML models.

    Standard encoding: Male=1, Female=0, Other/Unknown=None

    Args:
        gender: Gender string (case-insensitive)

    Returns:
        Encoded gender (1 for male, 0 for female, None for other/unknown)
    """
    if not gender:
        return None

    gender_lower = gender.lower()
    if gender_lower in ["male", "m"]:
        return 1
    elif gender_lower in ["female", "f"]:
        return 0
    else:
        return None