diff --git a/docs/reference/utilities/fhir_helpers.md b/docs/reference/utilities/fhir_helpers.md index 35d531ad..40cedbca 100644 --- a/docs/reference/utilities/fhir_helpers.md +++ b/docs/reference/utilities/fhir_helpers.md @@ -2,6 +2,98 @@ The `fhir` module provides a set of helper functions to make it easier for you to work with FHIR resources. +## FHIR Version Support + +HealthChain supports multiple FHIR versions: **R5** (default), **R4B**, and **STU3**. All resource creation and helper functions accept an optional `version` parameter. + +### Supported Versions + +| Version | Description | Package Path | +|---------|-------------|--------------| +| **R5** | FHIR Release 5 (default) | `fhir.resources.*` | +| **R4B** | FHIR R4B (Ballot) | `fhir.resources.R4B.*` | +| **STU3** | FHIR STU3 | `fhir.resources.STU3.*` | + +### Basic Usage + +```python +from healthchain.fhir import ( + FHIRVersion, + get_fhir_resource, + set_default_version, + fhir_version_context, + convert_resource, + create_condition, +) + +# Get a resource class for a specific version +Patient_R4B = get_fhir_resource("Patient", "R4B") +Patient_R5 = get_fhir_resource("Patient", FHIRVersion.R5) + +# Create resources with a specific version +condition_r4b = create_condition( + subject="Patient/123", + code="38341003", + display="Hypertension", + version="R4B" # Creates R4B Condition +) + +# Set the default version for the session +set_default_version("R4B") + +# Use context manager for temporary version changes +with fhir_version_context("STU3"): + # All resources created here use STU3 + condition = create_condition(subject="Patient/123", code="123") +``` + +### Version Conversion + +Convert resources between FHIR versions using `convert_resource()`: + +```python +from healthchain.fhir import get_fhir_resource, convert_resource + +# Create an R5 Patient +Patient_R5 = get_fhir_resource("Patient") +patient_r5 = Patient_R5(id="test-123", gender="male") + +# Convert to R4B +patient_r4b = convert_resource(patient_r5, "R4B") +print(patient_r4b.__class__.__module__) # fhir.resources.R4B.patient +``` + +!!! warning "Version Conversion Limitations" + The `convert_resource()` function uses a serialize/deserialize approach. Field mappings between FHIR versions may not be 1:1 - some fields may be added, removed, or renamed between versions. Complex resources with version-specific fields may require manual handling. + +### Version Detection + +Detect the FHIR version of an existing resource: + +```python +from healthchain.fhir import get_resource_version, get_fhir_resource + +Patient_R4B = get_fhir_resource("Patient", "R4B") +patient = Patient_R4B(id="123") + +version = get_resource_version(patient) +print(version) # FHIRVersion.R4B +``` + +### API Reference + +| Function | Description | +|----------|-------------| +| `get_fhir_resource(name, version)` | Get a resource class for a specific version | +| `get_default_version()` | Get the current default FHIR version | +| `set_default_version(version)` | Set the global default FHIR version | +| `reset_default_version()` | Reset to library default (R5) | +| `fhir_version_context(version)` | Context manager for temporary version changes | +| `convert_resource(resource, version)` | Convert a resource to a different version | +| `get_resource_version(resource)` | Detect the version of an existing resource | + +--- + ## Resource Creation FHIR is the modern de facto standard for storing and exchanging healthcare data, but working with [FHIR resources](https://www.hl7.org/fhir/resourcelist.html) can often involve complex and nested JSON structures with required and optional fields that vary between contexts. @@ -80,6 +172,14 @@ condition = create_condition( system="http://snomed.info/sct", ) +# Create an R4B condition +condition_r4b = create_condition( + subject="Patient/123", + code="38341003", + display="Hypertension", + version="R4B", # Optional: specify FHIR version +) + # Output the created resource print(condition.model_dump()) ``` diff --git a/healthchain/configs/defaults.yaml b/healthchain/configs/defaults.yaml index 16e33509..861704b4 100644 --- a/healthchain/configs/defaults.yaml +++ b/healthchain/configs/defaults.yaml @@ -1,6 +1,10 @@ # HealthChain Interoperability Engine Default Configuration # This file contains default values used throughout the engine +# FHIR version configuration +fhir: + version: "R5" # Default FHIR version: R5, R4B, or STU3 + defaults: # Common defaults for all resources common: diff --git a/healthchain/fhir/__init__.py b/healthchain/fhir/__init__.py index b33116d2..da59f240 100644 --- a/healthchain/fhir/__init__.py +++ b/healthchain/fhir/__init__.py @@ -1,5 +1,16 @@ """FHIR utilities for HealthChain.""" +from healthchain.fhir.version import ( + FHIRVersion, + get_fhir_resource, + get_default_version, + set_default_version, + reset_default_version, + fhir_version_context, + convert_resource, + get_resource_version, +) + from healthchain.fhir.resourcehelpers import ( create_condition, create_medication_statement, @@ -30,6 +41,7 @@ from healthchain.fhir.bundlehelpers import ( create_bundle, add_resource, + get_resource_type, get_resources, set_resources, merge_bundles, @@ -52,6 +64,15 @@ ) __all__ = [ + # Version management + "FHIRVersion", + "get_fhir_resource", + "get_default_version", + "set_default_version", + "reset_default_version", + "fhir_version_context", + "convert_resource", + "get_resource_version", # Resource creation "create_condition", "create_medication_statement", @@ -60,13 +81,10 @@ "create_patient", "create_risk_assessment_from_prediction", "create_document_reference", + "create_document_reference_content", # Element creation "create_single_codeable_concept", "create_single_reaction", - "set_condition_category", - "read_content_attachment", - "create_document_reference", - "create_document_reference_content", "create_single_attachment", # Resource modification "set_condition_category", @@ -80,6 +98,7 @@ # Bundle operations "create_bundle", "add_resource", + "get_resource_type", "get_resources", "set_resources", "merge_bundles", diff --git a/healthchain/fhir/bundlehelpers.py b/healthchain/fhir/bundlehelpers.py index 3edbc7ed..2bfbda7b 100644 --- a/healthchain/fhir/bundlehelpers.py +++ b/healthchain/fhir/bundlehelpers.py @@ -8,10 +8,12 @@ - extract_*(): extract resources from a bundle """ -from typing import List, Type, TypeVar, Optional, Union +from typing import List, Type, TypeVar, Optional, Union, TYPE_CHECKING from fhir.resources.bundle import Bundle, BundleEntry from fhir.resources.resource import Resource +if TYPE_CHECKING: + from healthchain.fhir.version import FHIRVersion T = TypeVar("T", bound=Resource) @@ -44,14 +46,19 @@ def add_resource( bundle.entry = (bundle.entry or []) + [entry] -def get_resource_type(resource_type: Union[str, Type[Resource]]) -> Type[Resource]: +def get_resource_type( + resource_type: Union[str, Type[Resource]], + version: Optional[Union["FHIRVersion", str]] = None, +) -> 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 + version: Optional FHIR version (e.g., "R4B", "STU3", or FHIRVersion enum). + If None, uses the current default version. Returns: - The resource type class + The resource type class for the specified version Raises: ValueError: If the resource type is not supported or cannot be imported @@ -64,17 +71,10 @@ def get_resource_type(resource_type: Union[str, Type[Resource]]) -> Type[Resourc 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 + # Use version manager for dynamic import with version support + from healthchain.fhir.version import get_fhir_resource + + return get_fhir_resource(resource_type, version) def get_resources( diff --git a/healthchain/fhir/elementhelpers.py b/healthchain/fhir/elementhelpers.py index c4b4532f..9f86811f 100644 --- a/healthchain/fhir/elementhelpers.py +++ b/healthchain/fhir/elementhelpers.py @@ -8,11 +8,10 @@ import base64 import datetime -from typing import Optional, List, Dict, Any -from fhir.resources.codeableconcept import CodeableConcept -from fhir.resources.codeablereference import CodeableReference -from fhir.resources.coding import Coding -from fhir.resources.attachment import Attachment +from typing import Optional, List, Dict, Any, Union, TYPE_CHECKING + +if TYPE_CHECKING: + from healthchain.fhir.version import FHIRVersion logger = logging.getLogger(__name__) @@ -21,7 +20,8 @@ def create_single_codeable_concept( code: str, display: Optional[str] = None, system: Optional[str] = "http://snomed.info/sct", -) -> CodeableConcept: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """ Create a minimal FHIR CodeableConcept with a single coding. @@ -29,10 +29,16 @@ def create_single_codeable_concept( code: REQUIRED. The code value from the code system display: The display name for the code system: The code system (default: SNOMED CT) + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. Returns: CodeableConcept: A FHIR CodeableConcept resource with a single coding """ + from healthchain.fhir.version import get_fhir_resource + + CodeableConcept = get_fhir_resource("CodeableConcept", version) + Coding = get_fhir_resource("Coding", version) + return CodeableConcept(coding=[Coding(system=system, code=code, display=display)]) @@ -41,6 +47,7 @@ def create_single_reaction( display: Optional[str] = None, system: Optional[str] = "http://snomed.info/sct", severity: Optional[str] = None, + version: Optional[Union["FHIRVersion", str]] = None, ) -> List[Dict[str, Any]]: """Create a minimal FHIR Reaction with a single coding. @@ -53,10 +60,17 @@ def create_single_reaction( 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) + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. Returns: A list containing a single FHIR Reaction dictionary with manifestation and severity fields """ + from healthchain.fhir.version import get_fhir_resource + + CodeableConcept = get_fhir_resource("CodeableConcept", version) + CodeableReference = get_fhir_resource("CodeableReference", version) + Coding = get_fhir_resource("Coding", version) + return [ { "manifestation": [ @@ -76,7 +90,8 @@ def create_single_attachment( data: Optional[str] = None, url: Optional[str] = None, title: Optional[str] = "Attachment created by HealthChain", -) -> Attachment: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """Create a minimal FHIR Attachment. Creates a FHIR Attachment resource with basic fields. Either data or url should be provided. @@ -87,10 +102,14 @@ def create_single_attachment( 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") + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. Returns: Attachment: A FHIR Attachment resource with basic metadata and content """ + from healthchain.fhir.version import get_fhir_resource + + Attachment = get_fhir_resource("Attachment", version) if not data and not url: logger.warning("No data or url provided for attachment") diff --git a/healthchain/fhir/resourcehelpers.py b/healthchain/fhir/resourcehelpers.py index bb278cef..dcac457e 100644 --- a/healthchain/fhir/resourcehelpers.py +++ b/healthchain/fhir/resourcehelpers.py @@ -13,21 +13,12 @@ import logging import datetime -from typing import List, Optional, Dict, Any -from fhir.resources.coding import Coding -from fhir.resources.condition import Condition -from fhir.resources.identifier import Identifier -from fhir.resources.medicationstatement import MedicationStatement -from fhir.resources.allergyintolerance import AllergyIntolerance -from fhir.resources.documentreference import DocumentReference -from fhir.resources.observation import Observation -from fhir.resources.resource import Resource -from fhir.resources.riskassessment import RiskAssessment -from fhir.resources.patient import Patient -from fhir.resources.quantity import Quantity +from typing import List, Optional, Dict, Any, Union, TYPE_CHECKING + +# Keep static imports only for types that are always version-compatible +# and used in signatures/type hints from fhir.resources.codeableconcept import CodeableConcept from fhir.resources.reference import Reference -from fhir.resources.meta import Meta from healthchain.fhir.elementhelpers import ( create_single_codeable_concept, @@ -35,6 +26,9 @@ ) from healthchain.fhir.utilities import _generate_id +if TYPE_CHECKING: + from healthchain.fhir.version import FHIRVersion + logger = logging.getLogger(__name__) @@ -44,7 +38,8 @@ def create_condition( code: Optional[str] = None, display: Optional[str] = None, system: Optional[str] = "http://snomed.info/sct", -) -> Condition: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """ Create a minimal active FHIR Condition. If you need to create a more complex condition, use the FHIR Condition resource directly. @@ -56,22 +51,29 @@ def create_condition( code: The condition code display: The display name for the condition system: The code system (default: SNOMED CT) + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. Returns: Condition: A FHIR Condition resource with an auto-generated ID prefixed with 'hc-' """ + from healthchain.fhir.version import get_fhir_resource + + Condition = get_fhir_resource("Condition", version) + ReferenceClass = get_fhir_resource("Reference", version) + if code: - condition_code = create_single_codeable_concept(code, display, system) + condition_code = create_single_codeable_concept(code, display, system, version) else: condition_code = None condition = Condition( id=_generate_id(), - subject=Reference(reference=subject), + subject=ReferenceClass(reference=subject), clinicalStatus=create_single_codeable_concept( code=clinical_status, display=clinical_status.capitalize(), system="http://terminology.hl7.org/CodeSystem/condition-clinical", + version=version, ), code=condition_code, ) @@ -85,7 +87,8 @@ def create_medication_statement( code: Optional[str] = None, display: Optional[str] = None, system: Optional[str] = "http://snomed.info/sct", -) -> MedicationStatement: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """ Create a minimal recorded FHIR MedicationStatement. If you need to create a more complex medication statement, use the FHIR MedicationStatement resource directly. @@ -97,18 +100,26 @@ def create_medication_statement( code: The medication code display: The display name for the medication system: The code system (default: SNOMED CT) + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. Returns: MedicationStatement: A FHIR MedicationStatement resource with an auto-generated ID prefixed with 'hc-' """ + from healthchain.fhir.version import get_fhir_resource + + MedicationStatement = get_fhir_resource("MedicationStatement", version) + ReferenceClass = get_fhir_resource("Reference", version) + if code: - medication_concept = create_single_codeable_concept(code, display, system) + medication_concept = create_single_codeable_concept( + code, display, system, version + ) else: medication_concept = None medication = MedicationStatement( id=_generate_id(), - subject=Reference(reference=subject), + subject=ReferenceClass(reference=subject), status=status, medication={"concept": medication_concept}, ) @@ -121,7 +132,8 @@ def create_allergy_intolerance( code: Optional[str] = None, display: Optional[str] = None, system: Optional[str] = "http://snomed.info/sct", -) -> AllergyIntolerance: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """ Create a minimal active FHIR AllergyIntolerance. If you need to create a more complex allergy intolerance, use the FHIR AllergyIntolerance resource directly. @@ -132,18 +144,24 @@ def create_allergy_intolerance( code: The allergen code display: The display name for the allergen system: The code system (default: SNOMED CT) + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. Returns: AllergyIntolerance: A FHIR AllergyIntolerance resource with an auto-generated ID prefixed with 'hc-' """ + from healthchain.fhir.version import get_fhir_resource + + AllergyIntolerance = get_fhir_resource("AllergyIntolerance", version) + ReferenceClass = get_fhir_resource("Reference", version) + if code: - allergy_code = create_single_codeable_concept(code, display, system) + allergy_code = create_single_codeable_concept(code, display, system, version) else: allergy_code = None allergy = AllergyIntolerance( id=_generate_id(), - patient=Reference(reference=patient), + patient=ReferenceClass(reference=patient), code=allergy_code, ) @@ -159,7 +177,8 @@ def create_value_quantity_observation( system: str = "http://loinc.org", display: Optional[str] = None, effective_datetime: Optional[str] = None, -) -> Observation: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """ 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. @@ -174,22 +193,29 @@ def create_value_quantity_observation( 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") + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. Returns: Observation: A FHIR Observation resource with an auto-generated ID prefixed with 'hc-' """ + from healthchain.fhir.version import get_fhir_resource + + Observation = get_fhir_resource("Observation", version) + ReferenceClass = get_fhir_resource("Reference", version) + Quantity = get_fhir_resource("Quantity", version) + 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) + subject_ref = ReferenceClass(reference=subject) observation = Observation( id=_generate_id(), status=status, - code=create_single_codeable_concept(code, display, system), + code=create_single_codeable_concept(code, display, system, version), subject=subject_ref, effectiveDateTime=effective_datetime, valueQuantity=Quantity( @@ -205,7 +231,8 @@ def create_patient( birth_date: Optional[str] = None, identifier: Optional[str] = None, identifier_system: Optional[str] = "http://hospital.example.org", -) -> Patient: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """ 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 @@ -216,13 +243,19 @@ def create_patient( 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") + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. Returns: Patient: A FHIR Patient resource with an auto-generated ID prefixed with 'hc-' """ + from healthchain.fhir.version import get_fhir_resource + + Patient = get_fhir_resource("Patient", version) + Identifier = get_fhir_resource("Identifier", version) + patient_id = _generate_id() - patient_data = {"id": patient_id} + patient_data: Dict[str, Any] = {"id": patient_id} if birth_date: patient_data["birthDate"] = birth_date @@ -250,7 +283,8 @@ def create_risk_assessment_from_prediction( basis: Optional[List[Reference]] = None, comment: Optional[str] = None, occurrence_datetime: Optional[str] = None, -) -> RiskAssessment: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """ 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. @@ -266,8 +300,8 @@ def create_risk_assessment_from_prediction( 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. + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. Returns: RiskAssessment: A FHIR RiskAssessment resource with an auto-generated ID prefixed with 'hc-' @@ -280,6 +314,11 @@ def create_risk_assessment_from_prediction( ... } >>> risk = create_risk_assessment("Patient/123", prediction) """ + from healthchain.fhir.version import get_fhir_resource + + RiskAssessment = get_fhir_resource("RiskAssessment", version) + ReferenceClass = get_fhir_resource("Reference", version) + if not occurrence_datetime: occurrence_datetime = datetime.datetime.now(datetime.timezone.utc).strftime( "%Y-%m-%dT%H:%M:%S%z" @@ -291,11 +330,12 @@ def create_risk_assessment_from_prediction( code=outcome["code"], display=outcome.get("display"), system=outcome.get("system", "http://snomed.info/sct"), + version=version, ) else: outcome_concept = outcome - prediction_data = { + prediction_data: Dict[str, Any] = { "outcome": outcome_concept, } @@ -307,12 +347,13 @@ def create_risk_assessment_from_prediction( code=prediction["qualitative_risk"], display=prediction["qualitative_risk"].capitalize(), system="http://terminology.hl7.org/CodeSystem/risk-probability", + version=version, ) - risk_assessment_data = { + risk_assessment_data: Dict[str, Any] = { "id": _generate_id(), "status": status, - "subject": Reference(reference=subject), + "subject": ReferenceClass(reference=subject), "occurrenceDateTime": occurrence_datetime, "prediction": [prediction_data], } @@ -338,7 +379,8 @@ def create_document_reference( status: str = "current", description: Optional[str] = "DocumentReference created by HealthChain", attachment_title: Optional[str] = "Attachment created by HealthChain", -) -> DocumentReference: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """ Create a minimal FHIR DocumentReference. If you need to create a more complex document reference, use the FHIR DocumentReference resource directly. @@ -351,10 +393,15 @@ def create_document_reference( status: REQUIRED. Status of the document reference (default: current) description: Description of the document reference attachment_title: Title for the document attachment + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. Returns: DocumentReference: A FHIR DocumentReference resource with an auto-generated ID prefixed with 'hc-' """ + from healthchain.fhir.version import get_fhir_resource + + DocumentReference = get_fhir_resource("DocumentReference", version) + document_reference = DocumentReference( id=_generate_id(), status=status, @@ -369,6 +416,7 @@ def create_document_reference( data=data, url=url, title=attachment_title, + version=version, ) } ], @@ -383,6 +431,7 @@ def create_document_reference_content( content_type: str = "text/plain", language: Optional[str] = "en-US", title: Optional[str] = None, + version: Optional[Union["FHIRVersion", str]] = None, **kwargs, ) -> Dict[str, Any]: """Create a FHIR DocumentReferenceContent object. @@ -397,6 +446,7 @@ def create_document_reference_content( 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") + version: FHIR version to use (e.g., "R4B", "STU3"). Defaults to current default. **kwargs: Additional DocumentReferenceContent fields (e.g., format, profile) Returns: @@ -437,6 +487,7 @@ def create_document_reference_content( data=attachment_data, url=url, title=title, + version=version, ) content: Dict[str, Any] = { @@ -451,13 +502,18 @@ def create_document_reference_content( return content -def set_condition_category(condition: Condition, category: str) -> Condition: +def set_condition_category( + condition: Any, + category: str, + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """ 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'. + version: FHIR version to use. If None, attempts to detect from the condition resource. Returns: Condition: The modified FHIR Condition resource with the specified category set @@ -465,6 +521,12 @@ def set_condition_category(condition: Condition, category: str) -> Condition: Raises: ValueError: If the category is not one of the allowed values. """ + from healthchain.fhir.version import get_resource_version + + # Detect version from resource if not provided + if version is None: + version = get_resource_version(condition) + allowed_categories = { "problem-list-item": { "code": "problem-list-item", @@ -486,17 +548,19 @@ def set_condition_category(condition: Condition, category: str) -> Condition: code=cat_info["code"], display=cat_info["display"], system="http://terminology.hl7.org/CodeSystem/condition-category", + version=version, ) ] return condition def add_provenance_metadata( - resource: Resource, + resource: Any, source: str, tag_code: Optional[str] = None, tag_display: Optional[str] = None, -) -> Resource: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """Add provenance metadata to a FHIR resource. Adds source system identifier, timestamp, and optional processing tags to track @@ -507,6 +571,7 @@ def add_provenance_metadata( 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 + version: FHIR version to use. If None, attempts to detect from the resource. Returns: Resource: The resource with added provenance metadata @@ -515,6 +580,15 @@ def add_provenance_metadata( >>> condition = create_condition(subject="Patient/123", code="E11.9") >>> condition = add_provenance_metadata(condition, "epic", "aggregated", "Aggregated from source") """ + from healthchain.fhir.version import get_fhir_resource, get_resource_version + + # Detect version from resource if not provided + if version is None: + version = get_resource_version(resource) + + Meta = get_fhir_resource("Meta", version) + Coding = get_fhir_resource("Coding", version) + if not resource.meta: resource.meta = Meta() @@ -541,11 +615,12 @@ def add_provenance_metadata( def add_coding_to_codeable_concept( - codeable_concept: CodeableConcept, + codeable_concept: Any, code: str, system: str, display: Optional[str] = None, -) -> CodeableConcept: + version: Optional[Union["FHIRVersion", str]] = None, +) -> Any: """Add a coding to an existing CodeableConcept. Useful for adding standardized codes (e.g., SNOMED CT) to resources that already @@ -556,6 +631,7 @@ def add_coding_to_codeable_concept( code: The code value from the code system system: The code system URI display: Optional display text for the code + version: FHIR version to use. If None, attempts to detect from the CodeableConcept. Returns: CodeableConcept: The updated CodeableConcept with the new coding added @@ -570,6 +646,14 @@ def add_coding_to_codeable_concept( ... display="Type 2 diabetes mellitus" ... ) """ + from healthchain.fhir.version import get_fhir_resource, get_resource_version + + # Detect version from CodeableConcept if not provided + if version is None: + version = get_resource_version(codeable_concept) + + Coding = get_fhir_resource("Coding", version) + if not codeable_concept.coding: codeable_concept.coding = [] diff --git a/healthchain/fhir/version.py b/healthchain/fhir/version.py new file mode 100644 index 00000000..19dd314e --- /dev/null +++ b/healthchain/fhir/version.py @@ -0,0 +1,254 @@ +"""FHIR version management for multi-version support. + +This module provides utilities for working with different FHIR versions (STU3, R4B, R5). +It enables dynamic resource loading, version context management, and basic resource conversion. + +Usage: + from healthchain.fhir.version import get_fhir_resource, FHIRVersion + + # Get a resource class for a specific version + Patient_R4B = get_fhir_resource("Patient", "R4B") + Patient_R5 = get_fhir_resource("Patient", FHIRVersion.R5) + + # Set the default version for the session + set_default_version("R4B") + + # Use context manager for temporary version changes + with fhir_version_context("STU3"): + patient = get_fhir_resource("Patient") # Returns STU3 Patient + + # Convert resources between versions + patient_r5 = Patient(id="123", gender="male") + patient_r4b = convert_resource(patient_r5, "R4B") +""" + +import importlib +import logging +from contextlib import contextmanager +from enum import Enum +from typing import Any, Generator, Optional, Type, Union + +logger = logging.getLogger(__name__) + + +class FHIRVersion(str, Enum): + """Supported FHIR versions. + + R5 is the default version in fhir.resources library. + R4B and STU3 are available via subpackages (e.g., fhir.resources.R4B). + """ + + STU3 = "STU3" + R4B = "R4B" + R5 = "R5" + + +# Module-level default version (None means use R5) +_default_version: Optional[FHIRVersion] = None + + +def _resolve_version(version: Optional[Union[FHIRVersion, str]]) -> FHIRVersion: + """Resolve a version parameter to a FHIRVersion enum. + + Args: + version: Version as enum, string, or None for default + + Returns: + FHIRVersion enum value + + Raises: + ValueError: If version string is not a valid FHIR version + """ + if version is None: + return get_default_version() + + if isinstance(version, FHIRVersion): + return version + + try: + return FHIRVersion(version.upper()) + except ValueError: + valid_versions = [v.value for v in FHIRVersion] + raise ValueError( + f"Invalid FHIR version '{version}'. Must be one of: {valid_versions}" + ) + + +def get_fhir_resource( + resource_name: str, version: Optional[Union[FHIRVersion, str]] = None +) -> Type[Any]: + """Dynamically import a FHIR resource class based on version. + + Args: + resource_name: Name of the FHIR resource (e.g., "Patient", "Condition") + version: FHIR version (None for default, or FHIRVersion enum/string) + + Returns: + The FHIR resource class for the specified version + + Raises: + ValueError: If version is invalid or resource cannot be imported + + Example: + >>> Patient_R5 = get_fhir_resource("Patient") + >>> Patient_R4B = get_fhir_resource("Patient", "R4B") + >>> Patient_STU3 = get_fhir_resource("Patient", FHIRVersion.STU3) + """ + resolved_version = _resolve_version(version) + + # Build module path based on version + # R5 is the default (no subpackage), R4B and STU3 use subpackages + if resolved_version == FHIRVersion.R5: + module_path = f"fhir.resources.{resource_name.lower()}" + else: + module_path = f"fhir.resources.{resolved_version.value}.{resource_name.lower()}" + + try: + module = importlib.import_module(module_path) + resource_class = getattr(module, resource_name) + logger.debug(f"Loaded {resource_name} from {module_path}") + return resource_class + except ImportError as e: + raise ValueError( + f"Could not import resource type: {resource_name}. " + f"Make sure it is a valid FHIR resource type for version '{resolved_version.value}'." + ) from e + except AttributeError as e: + raise ValueError( + f"Module '{module_path}' does not contain resource '{resource_name}'." + ) from e + + +def get_default_version() -> FHIRVersion: + """Get the current default FHIR version. + + Returns: + The current default FHIRVersion (R5 if not explicitly set) + """ + return _default_version or FHIRVersion.R5 + + +def set_default_version(version: Union[FHIRVersion, str]) -> None: + """Set the global default FHIR version. + + Args: + version: The FHIR version to use as default + + Example: + >>> set_default_version("R4B") + >>> patient = get_fhir_resource("Patient") # Returns R4B Patient + """ + global _default_version + _default_version = _resolve_version(version) + logger.info(f"Default FHIR version set to {_default_version.value}") + + +def reset_default_version() -> None: + """Reset the default FHIR version to library default (R5).""" + global _default_version + _default_version = None + logger.debug("Default FHIR version reset to R5") + + +@contextmanager +def fhir_version_context( + version: Union[FHIRVersion, str], +) -> Generator[FHIRVersion, None, None]: + """Context manager for temporarily changing the default FHIR version. + + Args: + version: The FHIR version to use within the context + + Yields: + The resolved FHIRVersion being used + + Example: + >>> with fhir_version_context("R4B") as v: + ... patient = get_fhir_resource("Patient") # R4B Patient + ... print(f"Using {v}") + >>> # After context, default is restored + """ + global _default_version + previous_version = _default_version + resolved = _resolve_version(version) + _default_version = resolved + try: + yield resolved + finally: + _default_version = previous_version + + +def convert_resource(resource: Any, target_version: Union[FHIRVersion, str]) -> Any: + """Convert a FHIR resource to a different version. + + Converts by serializing the resource to a dictionary and deserializing + with the target version's resource class. This approach works for + resources with compatible field structures. + + Note: + Field mappings between FHIR versions may not be 1:1. Some fields + may be added, removed, or renamed between versions. This function + performs a best-effort conversion and may raise validation errors + if the resource data is incompatible with the target version. + + Args: + resource: The FHIR resource to convert + target_version: The target FHIR version + + Returns: + A new resource instance of the target version + + Raises: + ValueError: If the resource type cannot be determined or imported + ValidationError: If the resource data is incompatible with target version + + Example: + >>> from fhir.resources.patient import Patient + >>> patient_r5 = Patient(id="123", gender="male") + >>> patient_r4b = convert_resource(patient_r5, "R4B") + >>> print(patient_r4b.__class__.__module__) + fhir.resources.R4B.patient + """ + # Get the resource type name from the class + resource_type = resource.__class__.__name__ + + # Get the target version's resource class + target_class = get_fhir_resource(resource_type, target_version) + + # Serialize to dict and deserialize with target class + data = resource.model_dump(exclude_none=True) + + logger.debug( + f"Converting {resource_type} from {resource.__class__.__module__} " + f"to {target_class.__module__}" + ) + + return target_class.model_validate(data) + + +def get_resource_version(resource: Any) -> Optional[FHIRVersion]: + """Detect the FHIR version of a resource based on its module path. + + Args: + resource: A FHIR resource instance + + Returns: + The FHIRVersion if detectable, None otherwise + + Example: + >>> from fhir.resources.R4B.patient import Patient + >>> patient = Patient(id="123") + >>> version = get_resource_version(patient) + >>> print(version) + FHIRVersion.R4B + """ + module = resource.__class__.__module__ + + if ".R4B." in module: + return FHIRVersion.R4B + elif ".STU3." in module: + return FHIRVersion.STU3 + elif module.startswith("fhir.resources."): + return FHIRVersion.R5 + + return None diff --git a/tests/fhir/test_version.py b/tests/fhir/test_version.py new file mode 100644 index 00000000..352378e7 --- /dev/null +++ b/tests/fhir/test_version.py @@ -0,0 +1,307 @@ +"""Tests for FHIR version management functionality.""" + +import pytest + +from healthchain.fhir.version import ( + FHIRVersion, + get_fhir_resource, + get_default_version, + set_default_version, + reset_default_version, + fhir_version_context, + convert_resource, + get_resource_version, + _resolve_version, +) + + +def test_fhir_version_enum_values(): + """Test FHIRVersion enum has expected values.""" + assert FHIRVersion.STU3 == "STU3" + assert FHIRVersion.R4B == "R4B" + assert FHIRVersion.R5 == "R5" + + +def test_fhir_version_enum_from_string(): + """Test creating FHIRVersion from string.""" + assert FHIRVersion("STU3") == FHIRVersion.STU3 + assert FHIRVersion("R4B") == FHIRVersion.R4B + assert FHIRVersion("R5") == FHIRVersion.R5 + + +def test_resolve_version_with_none(): + """Test _resolve_version returns default when None.""" + reset_default_version() + assert _resolve_version(None) == FHIRVersion.R5 + + +def test_resolve_version_with_enum(): + """Test _resolve_version passes through enum.""" + assert _resolve_version(FHIRVersion.R4B) == FHIRVersion.R4B + + +def test_resolve_version_with_string(): + """Test _resolve_version converts string to enum.""" + assert _resolve_version("R4B") == FHIRVersion.R4B + assert _resolve_version("r4b") == FHIRVersion.R4B # case insensitive + + +def test_resolve_version_invalid_string(): + """Test _resolve_version raises for invalid string.""" + with pytest.raises(ValueError, match="Invalid FHIR version"): + _resolve_version("invalid") + + +def test_get_default_version_initial(): + """Test initial default version is R5.""" + reset_default_version() + assert get_default_version() == FHIRVersion.R5 + + +def test_set_default_version_with_enum(): + """Test setting default version with enum.""" + reset_default_version() + set_default_version(FHIRVersion.R4B) + assert get_default_version() == FHIRVersion.R4B + reset_default_version() + + +def test_set_default_version_with_string(): + """Test setting default version with string.""" + reset_default_version() + set_default_version("STU3") + assert get_default_version() == FHIRVersion.STU3 + reset_default_version() + + +def test_reset_default_version(): + """Test resetting default version to R5.""" + set_default_version(FHIRVersion.R4B) + reset_default_version() + assert get_default_version() == FHIRVersion.R5 + + +def test_get_fhir_resource_r5_default(): + """Test loading resource with default R5 version.""" + reset_default_version() + Patient = get_fhir_resource("Patient") + assert Patient.__module__ == "fhir.resources.patient" + + +def test_get_fhir_resource_r4b(): + """Test loading resource with R4B version.""" + Patient = get_fhir_resource("Patient", "R4B") + assert Patient.__module__ == "fhir.resources.R4B.patient" + + +def test_get_fhir_resource_r4b_enum(): + """Test loading resource with R4B enum.""" + Patient = get_fhir_resource("Patient", FHIRVersion.R4B) + assert Patient.__module__ == "fhir.resources.R4B.patient" + + +def test_get_fhir_resource_stu3(): + """Test loading resource with STU3 version.""" + Patient = get_fhir_resource("Patient", "STU3") + assert Patient.__module__ == "fhir.resources.STU3.patient" + + +def test_get_fhir_resource_multiple_types(): + """Test loading multiple resource types.""" + Condition = get_fhir_resource("Condition", "R4B") + Bundle = get_fhir_resource("Bundle", "R4B") + Observation = get_fhir_resource("Observation", "R4B") + + assert Condition.__module__ == "fhir.resources.R4B.condition" + assert Bundle.__module__ == "fhir.resources.R4B.bundle" + assert Observation.__module__ == "fhir.resources.R4B.observation" + + +def test_get_fhir_resource_invalid_type(): + """Test loading invalid resource type raises ValueError.""" + with pytest.raises(ValueError, match="Could not import resource type"): + get_fhir_resource("InvalidResourceType") + + +def test_get_fhir_resource_respects_default_version(): + """Test get_fhir_resource uses default version when not specified.""" + reset_default_version() + set_default_version("R4B") + + Patient = get_fhir_resource("Patient") + assert Patient.__module__ == "fhir.resources.R4B.patient" + + reset_default_version() + + +def test_fhir_version_context_basic(): + """Test fhir_version_context changes version temporarily.""" + reset_default_version() + assert get_default_version() == FHIRVersion.R5 + + with fhir_version_context("R4B") as v: + assert v == FHIRVersion.R4B + assert get_default_version() == FHIRVersion.R4B + Patient = get_fhir_resource("Patient") + assert Patient.__module__ == "fhir.resources.R4B.patient" + + assert get_default_version() == FHIRVersion.R5 + + +def test_fhir_version_context_restores_on_exception(): + """Test fhir_version_context restores version even on exception.""" + reset_default_version() + assert get_default_version() == FHIRVersion.R5 + + with pytest.raises(RuntimeError): + with fhir_version_context("R4B"): + assert get_default_version() == FHIRVersion.R4B + raise RuntimeError("Test exception") + + assert get_default_version() == FHIRVersion.R5 + + +def test_fhir_version_context_nested(): + """Test nested fhir_version_context restores correctly.""" + reset_default_version() + + with fhir_version_context("R4B"): + assert get_default_version() == FHIRVersion.R4B + + with fhir_version_context("STU3"): + assert get_default_version() == FHIRVersion.STU3 + + assert get_default_version() == FHIRVersion.R4B + + assert get_default_version() == FHIRVersion.R5 + + +def test_convert_resource_r5_to_r4b(): + """Test converting a resource from R5 to R4B.""" + Patient_R5 = get_fhir_resource("Patient", "R5") + patient_r5 = Patient_R5(id="test-123", gender="male") + + patient_r4b = convert_resource(patient_r5, "R4B") + + assert patient_r4b.__class__.__module__ == "fhir.resources.R4B.patient" + assert patient_r4b.id == "test-123" + assert patient_r4b.gender == "male" + + +def test_convert_resource_r4b_to_r5(): + """Test converting a resource from R4B to R5.""" + Patient_R4B = get_fhir_resource("Patient", "R4B") + patient_r4b = Patient_R4B(id="test-456", gender="female") + + patient_r5 = convert_resource(patient_r4b, "R5") + + assert patient_r5.__class__.__module__ == "fhir.resources.patient" + assert patient_r5.id == "test-456" + assert patient_r5.gender == "female" + + +def test_convert_resource_with_enum(): + """Test convert_resource accepts FHIRVersion enum.""" + Patient_R5 = get_fhir_resource("Patient", "R5") + patient = Patient_R5(id="test", gender="other") + + converted = convert_resource(patient, FHIRVersion.R4B) + assert converted.__class__.__module__ == "fhir.resources.R4B.patient" + + +def test_get_resource_version_r5(): + """Test detecting version from R5 resource.""" + Patient = get_fhir_resource("Patient", "R5") + patient = Patient(id="test") + + version = get_resource_version(patient) + assert version == FHIRVersion.R5 + + +def test_get_resource_version_r4b(): + """Test detecting version from R4B resource.""" + Patient = get_fhir_resource("Patient", "R4B") + patient = Patient(id="test") + + version = get_resource_version(patient) + assert version == FHIRVersion.R4B + + +def test_get_resource_version_stu3(): + """Test detecting version from STU3 resource.""" + Patient = get_fhir_resource("Patient", "STU3") + patient = Patient(id="test") + + version = get_resource_version(patient) + assert version == FHIRVersion.STU3 + + +def test_get_resource_version_unknown(): + """Test get_resource_version returns None for non-FHIR objects.""" + + class FakeResource: + pass + + fake = FakeResource() + version = get_resource_version(fake) + assert version is None + + +# Integration tests for versioned resource helpers + + +def test_create_condition_with_version(): + """Test create_condition with version parameter.""" + from healthchain.fhir import create_condition + + cond_r5 = create_condition("Patient/1", code="123", display="Test") + cond_r4b = create_condition("Patient/1", code="123", display="Test", version="R4B") + + assert cond_r5.__class__.__module__ == "fhir.resources.condition" + assert cond_r4b.__class__.__module__ == "fhir.resources.R4B.condition" + + +def test_create_patient_with_version(): + """Test create_patient with version parameter.""" + from healthchain.fhir import create_patient + + patient_r5 = create_patient(gender="male") + patient_r4b = create_patient(gender="female", version="R4B") + + assert patient_r5.__class__.__module__ == "fhir.resources.patient" + assert patient_r4b.__class__.__module__ == "fhir.resources.R4B.patient" + + +def test_create_observation_with_version(): + """Test create_value_quantity_observation with version parameter.""" + from healthchain.fhir import create_value_quantity_observation + + obs_r5 = create_value_quantity_observation(code="12345", value=98.6, unit="F") + obs_r4b = create_value_quantity_observation( + code="12345", value=98.6, unit="F", version="R4B" + ) + + assert obs_r5.__class__.__module__ == "fhir.resources.observation" + assert obs_r4b.__class__.__module__ == "fhir.resources.R4B.observation" + + +def test_get_resource_type_with_version(): + """Test get_resource_type with version parameter.""" + from healthchain.fhir import get_resource_type + + Condition_R5 = get_resource_type("Condition") + Condition_R4B = get_resource_type("Condition", version="R4B") + + assert Condition_R5.__module__ == "fhir.resources.condition" + assert Condition_R4B.__module__ == "fhir.resources.R4B.condition" + + +def test_create_single_codeable_concept_with_version(): + """Test create_single_codeable_concept with version parameter.""" + from healthchain.fhir.elementhelpers import create_single_codeable_concept + + cc_r5 = create_single_codeable_concept("123", "Test") + cc_r4b = create_single_codeable_concept("123", "Test", version="R4B") + + assert cc_r5.__class__.__module__ == "fhir.resources.codeableconcept" + assert cc_r4b.__class__.__module__ == "fhir.resources.R4B.codeableconcept"