Fix top level extension implementation

fhirflat/resources/encounter.py

@@ -4,19 +4,24 @@
 import orjson
 
 from ..flat2fhir import expand_concepts
-from .extensions import relativeTimingPhaseExtension
-from pydantic.v1 import Field
-from typing import TypeAlias, ClassVar
+
+from .extensions import relativePhase, timingPhase
+from .extension_types import relativePhaseType, timingPhaseType
+from pydantic.v1 import Field, validator
+from typing import TypeAlias, ClassVar, Union
+from fhir.resources import fhirtypes
 
 JsonString: TypeAlias = str
 
 
 class Encounter(_Encounter, FHIRFlatBase):
 
-    extension: relativeTimingPhaseExtension = Field(
+    extension: list[
+        Union[relativePhaseType, timingPhaseType, fhirtypes.ExtensionType]
+    ] = Field(
         None,
         alias="extension",
-        title="Additional content defined by implementations",
+        title="List of `Extension` items (represented as `dict` in JSON)",
         description=(
             """
             Contains the G.H 'eventTiming' and 'relativePhase' extensions, and allows
@@ -25,6 +30,8 @@ class Encounter(_Encounter, FHIRFlatBase):
         ),
         # if property is element of this resource.
         element_property=True,
+        # this trys to match the type of the object to each of the union types
+        union_mode="smart",
     )
 
     # attributes to exclude from the flat representation
@@ -42,6 +49,16 @@ class Encounter(_Encounter, FHIRFlatBase):
     # required attributes that are not present in the FHIRflat representation
     flat_defaults: ClassVar[list[str]] = FHIRFlatBase.flat_defaults + ["status"]
 
+    @validator("extension")
+    def validate_extension_contents(cls, extensions):
+        rel_phase_count = sum(isinstance(item, relativePhase) for item in extensions)
+        tim_phase_count = sum(isinstance(item, timingPhase) for item in extensions)
+
+        if rel_phase_count > 1 or tim_phase_count > 1:
+            raise ValueError("relativePhase and timingPhase can only appear once.")
+
+        return extensions
+
     @classmethod
     def cleanup(cls, data: JsonString) -> Encounter:
         """

fhirflat/resources/extension_types.py

@@ -47,17 +47,5 @@ class dateTimeExtensionType(AbstractType):
     __resource_type__ = "dateTimeExtension"
 
 
-class timingPhaseExtensionType(AbstractType):
-    __resource_type__ = "timingPhaseExtension"
-
-
-class relativePhaseExtensionType(AbstractType):
-    __resource_type__ = "relativePhaseExtension"
-
-
-class relativeTimingPhaseExtensionType(AbstractType):
-    __resource_type__ = "relativeTimingPhaseExtension"
-
-
-class procedureExtensionType(AbstractType):
-    __resource_type__ = "procedureExtension"
+class relativePeriodType(AbstractType):
+    __resource_type__ = "relativePeriod"

fhirflat/resources/extension_validators.py

@@ -61,10 +61,7 @@ def __init__(self):
             "approximateDate": (None, ".extensions"),
             "Duration": (None, ".extensions"),
             "dateTimeExtension": (None, ".extensions"),
-            "timingPhaseExtension": (None, ".extensions"),
-            "relativePhaseExtension": (None, ".extensions"),
-            "relativeTimingPhaseExtension": (None, ".extensions"),
-            "procedureExtension": (None, ".extensions"),
+            "relativePeriod": (None, ".extensions"),
         }
 
     def get_fhir_model_class(self, model_name: str) -> Type[FHIRAbstractModel]:
@@ -225,19 +222,5 @@ def datetimeextension_validator(v: Union[StrBytes, dict, Path, FHIRAbstractModel
     return Validators().fhir_model_validator("dateTimeExtension", v)
 
 
-def timingphaseextension_validator(v: Union[StrBytes, dict, Path, FHIRAbstractModel]):
-    return Validators().fhir_model_validator("timingPhaseExtension", v)
-
-
-def relativephaseextension_validator(v: Union[StrBytes, dict, Path, FHIRAbstractModel]):
-    return Validators().fhir_model_validator("relativePhaseExtension", v)
-
-
-def relativetimingphaseextension_validator(
-    v: Union[StrBytes, dict, Path, FHIRAbstractModel]
-):
-    return Validators().fhir_model_validator("relativeTimingPhaseExtension", v)
-
-
-def procedureextension_validator(v: Union[StrBytes, dict, Path, FHIRAbstractModel]):
-    return Validators().fhir_model_validator("procedureExtension", v)
+def relativeperiod_validator(v: Union[StrBytes, dict, Path, FHIRAbstractModel]):
+    return Validators().fhir_model_validator("relativePeriod", v)

fhirflat/resources/extensions.py

@@ -145,6 +145,8 @@ def elements_sequence(cls):
         ]
 
 
+# Should this not be a FHIRextension like down the bottom? (copy the 'Period' or 'Range'
+# datatype)
 class relativePhase(_DataType):
 
     resource_type = Field("relativePhase", const=True)
@@ -342,36 +344,6 @@ def validate_extension_contents(cls, extensions):
         return extensions
 
 
-class timingPhaseExtension(_FHIRPrimitiveExtension):
-    """
-    A G.Health specific extension to the FHIR dateTime type
-    Allows dates to be specified as either approximate, and/or number of days relative
-    to the current date.
-    """
-
-    resource_type = Field("timingPhaseExtension", const=True)
-
-    extension: list[Union[et.timingPhaseType, fhirtypes.ExtensionType]] = Field(
-        None,
-        alias="extension",
-        title="List of `Extension` items (represented as `dict` in JSON)",
-        description="Additional content defined by implementations",
-        # if property is element of this resource.
-        element_property=True,
-        # this trys to match the type of the object to each of the union types
-        union_mode="smart",
-    )
-
-    @validator("extension")
-    def validate_extension_contents(cls, extensions):
-        phase_count = sum(isinstance(item, timingPhase) for item in extensions)
-
-        if phase_count > 1:
-            raise ValueError("timingPhase can only appear once.")
-
-        return extensions
-
-
 class relativePhaseExtension(_FHIRPrimitiveExtension):
     """
     A G.Health specific extension to the FHIR dateTime type
@@ -400,69 +372,3 @@ def validate_extension_contents(cls, extensions):
             raise ValueError("relativePhase can only appear once.")
 
         return extensions
-
-
-class relativeTimingPhaseExtension(_FHIRPrimitiveExtension):
-    """
-    Contains both the relative timing (pre-admission, during admission etc) and the
-    relative phase (number of days since admission for the start and end of an event)
-    extensions.
-    """
-
-    resource_type = Field("relativeTimingPhaseExtension", const=True)
-
-    extension: list[
-        Union[et.relativePhaseType, et.timingPhaseType, fhirtypes.ExtensionType]
-    ] = Field(
-        None,
-        alias="extension",
-        title="List of `Extension` items (represented as `dict` in JSON)",
-        description="Additional content defined by implementations",
-        # if property is element of this resource.
-        element_property=True,
-        # this trys to match the type of the object to each of the union types
-        union_mode="smart",
-    )
-
-    @validator("extension")
-    def validate_extension_contents(cls, extensions):
-        rel_phase_count = sum(isinstance(item, relativePhase) for item in extensions)
-        tim_phase_count = sum(isinstance(item, timingPhase) for item in extensions)
-
-        if rel_phase_count > 1 or tim_phase_count > 1:
-            raise ValueError("relativePhase and timingPhase can only appear once.")
-
-        return extensions
-
-
-class procedureExtension(_FHIRPrimitiveExtension):
-    """
-    Contains both the relative timing (pre-admission, during admission etc) and the
-    relative phase (number of days since admission for the start and end of an event)
-    extensions.
-    """
-
-    resource_type = Field("procedureExtension", const=True)
-
-    extension: list[
-        Union[et.durationType, et.timingPhaseType, fhirtypes.ExtensionType]
-    ] = Field(
-        None,
-        alias="extension",
-        title="List of `Extension` items (represented as `dict` in JSON)",
-        description="Additional content defined by implementations",
-        # if property is element of this resource.
-        element_property=True,
-        # this trys to match the type of the object to each of the union types
-        union_mode="smart",
-    )
-
-    @validator("extension")
-    def validate_extension_contents(cls, extensions):
-        duration_count = sum(isinstance(item, Duration) for item in extensions)
-        tim_phase_count = sum(isinstance(item, timingPhase) for item in extensions)
-
-        if duration_count > 1 or tim_phase_count > 1:
-            raise ValueError("Duration and timingPhase can only appear once.")
-
-        return extensions

fhirflat/resources/immunization.py

@@ -1,32 +1,36 @@
 from __future__ import annotations
 from fhir.resources.immunization import Immunization as _Immunization
 from .base import FHIRFlatBase
-from .extensions import dateTimeExtension, timingPhaseExtension
-from pydantic.v1 import Field
+from .extensions import timingPhase
+from .extension_types import timingPhaseType, dateTimeExtensionType
+from pydantic.v1 import Field, validator
 import orjson
 
 from ..flat2fhir import expand_concepts
-from typing import TypeAlias, ClassVar
+from typing import TypeAlias, ClassVar, Union
+from fhir.resources import fhirtypes
 
 JsonString: TypeAlias = str
 
 
 class Immunization(_Immunization, FHIRFlatBase):
 
-    extension: timingPhaseExtension = Field(
+    extension: list[Union[timingPhaseType, fhirtypes.ExtensionType]] = Field(
         None,
         alias="extension",
-        title="Additional content defined by implementations",
+        title="List of `Extension` items (represented as `dict` in JSON)",
         description=(
             """
             Contains the G.H 'eventPhase' extension, and allows extensions from other
              implementations to be included."""
         ),
         # if property is element of this resource.
         element_property=True,
+        # this trys to match the type of the object to each of the union types
+        union_mode="smart",
     )
 
-    occurrenceDateTime__ext: dateTimeExtension = Field(
+    occurrenceDateTime__ext: dateTimeExtensionType = Field(
         None,
         alias="_occurrenceDateTime",
         title="Extension field for ``occurrenceDateTime``.",
@@ -51,6 +55,15 @@ class Immunization(_Immunization, FHIRFlatBase):
     # required attributes that are not present in the FHIRflat representation
     flat_defaults: ClassVar[list[str]] = FHIRFlatBase.flat_defaults + ["status"]
 
+    @validator("extension")
+    def validate_extension_contents(cls, extensions):
+        phase_count = sum(isinstance(item, timingPhase) for item in extensions)
+
+        if phase_count > 1:
+            raise ValueError("timingPhase can only appear once.")
+
+        return extensions
+
     @classmethod
     def cleanup(cls, data: JsonString) -> Immunization:
         """

fhirflat/resources/observation.py

@@ -3,12 +3,14 @@
 from fhir.resources.observation import ObservationComponent as _ObservationComponent
 
 from .base import FHIRFlatBase
-from .extension_types import dateTimeExtensionType, timingPhaseExtensionType
-from pydantic.v1 import Field
+from .extension_types import dateTimeExtensionType, timingPhaseType
+from .extensions import timingPhase
+from pydantic.v1 import Field, validator
 import orjson
+from fhir.resources import fhirtypes
 
 from ..flat2fhir import expand_concepts
-from typing import TypeAlias, ClassVar
+from typing import TypeAlias, ClassVar, Union
 
 JsonString: TypeAlias = str
 
@@ -27,17 +29,19 @@ class ObservationComponent(_ObservationComponent):
 
 class Observation(_Observation, FHIRFlatBase):
 
-    extension: timingPhaseExtensionType = Field(
+    extension: list[Union[timingPhaseType, fhirtypes.ExtensionType]] = Field(
         None,
         alias="extension",
-        title="Additional content defined by implementations",
+        title="List of `Extension` items (represented as `dict` in JSON)",
         description=(
             """
             Contains the G.H 'eventPhase' extension, and allows extensions from other
              implementations to be included."""
         ),
         # if property is element of this resource.
         element_property=True,
+        # this trys to match the type of the object to each of the union types
+        union_mode="smart",
     )
 
     effectiveDateTime__ext: dateTimeExtensionType = Field(
@@ -78,6 +82,15 @@ class Observation(_Observation, FHIRFlatBase):
     # required attributes that are not present in the FHIRflat representation
     flat_defaults: ClassVar[list[str]] = FHIRFlatBase.flat_defaults + ["status"]
 
+    @validator("extension")
+    def validate_extension_contents(cls, extensions):
+        phase_count = sum(isinstance(item, timingPhase) for item in extensions)
+
+        if phase_count > 1:
+            raise ValueError("timingPhase can only appear once.")
+
+        return extensions
+
     @classmethod
     def cleanup(cls, data: JsonString) -> Observation:
         """