Add fhir versions

healthchain/__init__.py

@@ -3,6 +3,7 @@
 
 from .utils.logger import add_handlers
 from .config.base import ConfigManager, ValidationLevel
+from .fhir.version import set_fhir_version, get_fhir_version
 
 
 # Enable deprecation warnings
@@ -14,7 +15,15 @@
 logger.setLevel(logging.INFO)
 
 # Export them at the top level
-__all__ = ["ConfigManager", "ValidationLevel", "api", "ehr", "sandbox"]
+__all__ = [
+    "ConfigManager",
+    "ValidationLevel",
+    "set_fhir_version",
+    "get_fhir_version",
+    "api",
+    "ehr",
+    "sandbox",
+]
 
 
 # Legacy import with warning

healthchain/configs/defaults.yaml

@@ -30,6 +30,10 @@ defaults:
               code: "UNK"
               display: "Unknown"
 
+# FHIR version configuration
+fhir:
+  version: "R4B"  # Default FHIR version (R4, R4B, or R5)
+
 # TODO: More control over settings
 # # Validation settings
 # validation:

healthchain/fhir/__init__.py

@@ -51,6 +51,13 @@
     encode_gender,
 )
 
+from healthchain.fhir.version import (
+    get_fhir_version,
+    set_fhir_version,
+    version_context,
+    get_resource_class,
+)
+
 __all__ = [
     # Resource creation
     "create_condition",
@@ -95,4 +102,9 @@
     "calculate_age_from_birthdate",
     "calculate_age_from_event_date",
     "encode_gender",
+    # Version management
+    "get_fhir_version",
+    "set_fhir_version",
+    "version_context",
+    "get_resource_class",
 ]

healthchain/fhir/bundlehelpers.py

@@ -8,15 +8,20 @@
 - extract_*(): extract resources from a bundle
 """
 
-from typing import List, Type, TypeVar, Optional, Union
-from fhir.resources.bundle import Bundle, BundleEntry
-from fhir.resources.resource import Resource
+from typing import TYPE_CHECKING, List, Type, TypeVar, Optional, Union
 
+# Import version manager for lazy resource loading
+from healthchain.fhir.version import get_resource_class
 
-T = TypeVar("T", bound=Resource)
+# Type hints using string annotations (lazy evaluation)
+if TYPE_CHECKING:
+    from fhir.resources.bundle import Bundle, BundleEntry
+    from fhir.resources.resource import Resource
 
+T = TypeVar("T", bound="Resource")
 
-def create_bundle(bundle_type: str = "collection") -> Bundle:
+
+def create_bundle(bundle_type: str = "collection") -> "Bundle":
     """Create an empty FHIR Bundle.
     https://www.hl7.org/fhir/bundle.html
 
@@ -25,11 +30,13 @@ def create_bundle(bundle_type: str = "collection") -> Bundle:
             Valid types: document, message, transaction, transaction-response,
             batch, batch-response, history, searchset, collection
     """
+    # Lazy import version-aware resource classes
+    Bundle = get_resource_class("Bundle")
     return Bundle(type=bundle_type, entry=[])
 
 
 def add_resource(
-    bundle: Bundle, resource: Resource, full_url: Optional[str] = None
+    bundle: "Bundle", resource: "Resource", full_url: Optional[str] = None
 ) -> None:
     """Add a resource to a bundle.
 
@@ -38,13 +45,23 @@ def add_resource(
         resource: The resource to add, e.g. Condition, MedicationStatement, AllergyIntolerance
         full_url: Optional full URL for the resource
     """
+    # BundleEntry is a nested type within the Bundle class, not a separate resource
+    # We need to import it from the same module as Bundle
+    Bundle = get_resource_class("Bundle")
+    # Import the module to get BundleEntry
+    import importlib
+
+    module_name = Bundle.__module__
+    bundle_module = importlib.import_module(module_name)
+    BundleEntry = getattr(bundle_module, "BundleEntry")
+
     entry = BundleEntry(resource=resource)
     if full_url:
         entry.fullUrl = full_url
     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"]]) -> Type["Resource"]:
     """Get the resource type class from string or type.
 
     Args:
@@ -65,21 +82,21 @@ def get_resource_type(resource_type: Union[str, Type[Resource]]) -> Type[Resourc
         )
 
     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:
+        # Use version manager for dynamic import with version awareness
+        from healthchain.fhir.version import get_resource_class
+
+        return get_resource_class(resource_type)
+    except ValueError as e:
+        # Re-raise with more context
         raise ValueError(
             f"Could not import resource type: {resource_type}. "
             "Make sure it is a valid FHIR resource type."
         ) from e
 
 
 def get_resources(
-    bundle: Bundle, resource_type: Union[str, Type[Resource]]
-) -> List[Resource]:
+    bundle: "Bundle", resource_type: Union[str, Type["Resource"]]
+) -> List["Resource"]:
     """Get all resources of a specific type from a bundle.
 
     Args:
@@ -109,9 +126,9 @@ def get_resources(
 
 
 def set_resources(
-    bundle: Bundle,
-    resources: List[Resource],
-    resource_type: Union[str, Type[Resource]],
+    bundle: "Bundle",
+    resources: List["Resource"],
+    resource_type: Union[str, Type["Resource"]],
     replace: bool = True,
 ) -> None:
     """Set resources of a specific type in the bundle.
@@ -157,11 +174,11 @@ def set_resources(
 
 
 def merge_bundles(
-    bundles: List[Bundle],
+    bundles: List["Bundle"],
     bundle_type: str = "collection",
     deduplicate: bool = False,
     dedupe_key: str = "id",
-) -> Bundle:
+) -> "Bundle":
     """Merge multiple FHIR bundles into a single bundle.
 
     Combines entries from multiple bundles while preserving resource metadata.
@@ -225,8 +242,8 @@ def merge_bundles(
 
 
 def extract_resources(
-    bundle: Bundle, resource_type: Union[str, Type[Resource]]
-) -> List[Resource]:
+    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)
@@ -258,7 +275,7 @@ def extract_resources(
     return extracted
 
 
-def count_resources(bundle: Bundle) -> dict[str, int]:
+def count_resources(bundle: "Bundle") -> dict[str, int]:
     """Count resources by type in a bundle.
 
     Args:

healthchain/fhir/elementhelpers.py

@@ -8,11 +8,15 @@
 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 TYPE_CHECKING, Optional, List, Dict, Any
+
+# Import version manager for lazy resource loading
+from healthchain.fhir.version import get_resource_class
+
+# Type hints using string annotations (lazy evaluation)
+if TYPE_CHECKING:
+    from fhir.resources.codeableconcept import CodeableConcept
+    from fhir.resources.attachment import Attachment
 
 logger = logging.getLogger(__name__)
 
@@ -21,7 +25,7 @@ def create_single_codeable_concept(
     code: str,
     display: Optional[str] = None,
     system: Optional[str] = "http://snomed.info/sct",
-) -> CodeableConcept:
+) -> "CodeableConcept":
     """
     Create a minimal FHIR CodeableConcept with a single coding.
 
@@ -33,6 +37,10 @@ def create_single_codeable_concept(
     Returns:
         CodeableConcept: A FHIR CodeableConcept resource with a single coding
     """
+    # Lazy import version-aware resource classes
+    CodeableConcept = get_resource_class("CodeableConcept")
+    Coding = get_resource_class("Coding")
+
     return CodeableConcept(coding=[Coding(system=system, code=code, display=display)])
 
 
@@ -57,6 +65,11 @@ def create_single_reaction(
     Returns:
         A list containing a single FHIR Reaction dictionary with manifestation and severity fields
     """
+    # Lazy import version-aware resource classes
+    CodeableReference = get_resource_class("CodeableReference")
+    CodeableConcept = get_resource_class("CodeableConcept")
+    Coding = get_resource_class("Coding")
+
     return [
         {
             "manifestation": [
@@ -76,7 +89,7 @@ def create_single_attachment(
     data: Optional[str] = None,
     url: Optional[str] = None,
     title: Optional[str] = "Attachment created by HealthChain",
-) -> Attachment:
+) -> "Attachment":
     """Create a minimal FHIR Attachment.
 
     Creates a FHIR Attachment resource with basic fields. Either data or url should be provided.
@@ -91,6 +104,8 @@ def create_single_attachment(
     Returns:
         Attachment: A FHIR Attachment resource with basic metadata and content
     """
+    # Lazy import version-aware resource classes
+    Attachment = get_resource_class("Attachment")
 
     if not data and not url:
         logger.warning("No data or url provided for attachment")

healthchain/fhir/readers.py

@@ -5,18 +5,23 @@
 """
 
 import logging
-import importlib
 
-from typing import Optional, Dict, Any, List
-from fhir.resources.resource import Resource
-from fhir.resources.documentreference import DocumentReference
+from typing import TYPE_CHECKING, Optional, Dict, Any, List
+
+# Import version manager for lazy resource loading
+from healthchain.fhir.version import get_resource_class
+
+# Type hints using string annotations (lazy evaluation)
+if TYPE_CHECKING:
+    from fhir.resources.resource import Resource
+    from fhir.resources.documentreference import DocumentReference
 
 logger = logging.getLogger(__name__)
 
 
 def create_resource_from_dict(
     resource_dict: Dict, resource_type: str
-) -> Optional[Resource]:
+) -> Optional["Resource"]:
     """Create a FHIR resource instance from a dictionary
 
     Args:
@@ -27,10 +32,9 @@ def create_resource_from_dict(
         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)
+        from healthchain.fhir.version import get_resource_class
+
+        resource_class = get_resource_class(resource_type)
         return resource_class(**resource_dict)
     except Exception as e:
         logger.error(f"Failed to create FHIR resource: {str(e)}")
@@ -67,7 +71,7 @@ def prefetch_to_bundle(prefetch: Dict[str, Any]) -> Dict[str, Any]:
 
 def convert_prefetch_to_fhir_objects(
     prefetch_dict: Dict[str, Any],
-) -> Dict[str, Resource]:
+) -> 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
@@ -89,7 +93,6 @@ def convert_prefetch_to_fhir_objects(
         >>> isinstance(fhir_objects["patient"], Patient)  # True
         >>> isinstance(fhir_objects["condition"], Condition)  # True
     """
-    from fhir.resources import get_fhir_model_class
 
     result: Dict[str, Resource] = {}
 
@@ -99,7 +102,7 @@ def convert_prefetch_to_fhir_objects(
             resource_type = resource_data.get("resourceType")
             if resource_type:
                 try:
-                    resource_class = get_fhir_model_class(resource_type)
+                    resource_class = get_resource_class(resource_type)
                     result[key] = resource_class(**resource_data)
                 except Exception as e:
                     logger.warning(
@@ -122,7 +125,7 @@ def convert_prefetch_to_fhir_objects(
 
 
 def read_content_attachment(
-    document_reference: DocumentReference,
+    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.