Adds an Age and birthSex extension to the Patient resource (#22)

fhirflat/resources/extension_types.py

@@ -43,5 +43,13 @@ class durationType(AbstractType):
     __resource_type__ = "Duration"
 
 
+class ageType(AbstractType):
+    __resource_type__ = "Age"
+
+
+class birthSexType(AbstractType):
+    __resource_type__ = "birthSex"
+
+
 class dateTimeExtensionType(AbstractType):
     __resource_type__ = "dateTimeExtension"

fhirflat/resources/extension_validators.py

@@ -49,6 +49,9 @@
 if typing.TYPE_CHECKING:
     from pydantic.v1 import BaseModel
 
+# TODO: Make the validation error clearer when the error is that a
+# valueCodeableConcept is missing the outer "coding":[] list.
+
 
 class Validators:
     def __init__(self):
@@ -60,6 +63,8 @@ def __init__(self):
             "relativePeriod": (None, ".extensions"),
             "approximateDate": (None, ".extensions"),
             "Duration": (None, ".extensions"),
+            "Age": (None, ".extensions"),
+            "birthSex": (None, ".extensions"),
             "dateTimeExtension": (None, ".extensions"),
         }
 
@@ -217,5 +222,13 @@ def duration_validator(v: Union[StrBytes, dict, Path, FHIRAbstractModel]):
     return Validators().fhir_model_validator("Duration", v)
 
 
+def age_validator(v: Union[StrBytes, dict, Path, FHIRAbstractModel]):
+    return Validators().fhir_model_validator("Age", v)
+
+
+def birthsex_validator(v: Union[StrBytes, dict, Path, FHIRAbstractModel]):
+    return Validators().fhir_model_validator("birthSex", v)
+
+
 def datetimeextension_validator(v: Union[StrBytes, dict, Path, FHIRAbstractModel]):
     return Validators().fhir_model_validator("dateTimeExtension", v)

fhirflat/resources/extensions.py

@@ -342,6 +342,78 @@ def elements_sequence(cls):
         ]
 
 
+class Age(_DataType):
+    """
+    An ISARIC extension collecting data on the age of a patient.
+    """
+
+    resource_type = Field("Age", const=True)
+
+    url = Field("Age", const=True, alias="url")
+
+    valueQuantity: fhirtypes.QuantityType = Field(
+        None,
+        alias="valueQuantity",
+        title="Value of extension",
+        description=(
+            "Value of extension - must be one of a constrained set of the data "
+            "types (see [Extensibility](extensibility.html) for a list)."
+        ),
+        # if property is element of this resource.
+        element_property=True,
+        element_required=True,
+    )
+
+    @classmethod
+    def elements_sequence(cls):
+        """returning all elements names from
+        ``Extension`` according specification,
+        with preserving original sequence order.
+        """
+        return [
+            "id",
+            "extension",
+            "url",
+            "valueQuantity",
+        ]
+
+
+class birthSex(_DataType):
+    """
+    An ISARIC extension collecting data on the birth sex of a patient.
+    """
+
+    resource_type = Field("birthSex", const=True)
+
+    url = Field("birthSex", const=True, alias="url")
+
+    valueCodeableConcept: fhirtypes.CodeableConceptType = Field(
+        None,
+        alias="valueCodeableConcept",
+        title="Value of extension",
+        description=(
+            "Value of extension - must be one of a constrained set of the data "
+            "types (see [Extensibility](extensibility.html) for a list)."
+        ),
+        # if property is element of this resource.
+        element_property=True,
+        element_required=True,
+    )
+
+    @classmethod
+    def elements_sequence(cls):
+        """returning all elements names from
+        ``Extension`` according specification,
+        with preserving original sequence order.
+        """
+        return [
+            "id",
+            "extension",
+            "url",
+            "valueCodeableConcept",
+        ]
+
+
 # ------------------- extension types ------------------------------
 
 

fhirflat/resources/patient.py

@@ -1,12 +1,35 @@
 from fhir.resources.patient import Patient
 from .base import FHIRFlatBase
+from .extension_types import (
+    ageType,
+    birthSexType,
+)
+from .extensions import Age, birthSex
 import orjson
-from typing import TypeAlias, ClassVar
+
+from ..flat2fhir import expand_concepts
+from typing import TypeAlias, ClassVar, Union
+from fhir.resources import fhirtypes
+from pydantic.v1 import Field, validator
 
 JsonString: TypeAlias = str
 
 
 class Patient(Patient, FHIRFlatBase):
+    extension: list[Union[ageType, birthSexType, fhirtypes.ExtensionType]] = Field(
+        None,
+        alias="extension",
+        title="Additional content defined by implementations",
+        description=(
+            """
+            Contains the G.H 'Age' and 'birthSex' extensions,
+            and allows extensions from other implementations to be included."""
+        ),
+        # if property is element of this resource.
+        element_property=True,
+        union_mode="smart",
+    )
+
     # attributes to exclude from the flat representation
     flat_exclusions: ClassVar[set[str]] = FHIRFlatBase.flat_exclusions + (
         "identifier",
@@ -20,6 +43,16 @@ class Patient(Patient, FHIRFlatBase):
         "link",
     )
 
+    @validator("extension")
+    def validate_extension_contents(cls, extensions):
+        age_count = sum(isinstance(item, Age) for item in extensions)
+        birthsex_count = sum(isinstance(item, birthSex) for item in extensions)
+
+        if age_count > 1 or birthsex_count > 1:
+            raise ValueError("Age and birthSex can only appear once.")
+
+        return extensions
+
     @classmethod
     def flat_descriptions(cls) -> dict[str, str]:
         "Descriptions of the fields in the FHIRflat representation"
@@ -40,7 +73,15 @@ def cleanup(cls, data: JsonString) -> Patient:
         # Load the data and apply resource-specific changes
         data = orjson.loads(data)
 
-        # Strip time from the birthDate
-        data["birthDate"] = data["birthDate"].split("T", 1)[0]
+        # # Strip time from the birthDate
+        if "birthDate" in data:
+            data["birthDate"] = data["birthDate"].split("T", 1)[0]
+
+        data = expand_concepts(data, cls)
+
+        # create lists for properties which are lists of FHIR types
+        for field in [x for x in data.keys() if x in cls.attr_lists()]:
+            if type(data[field]) is not list:
+                data[field] = [data[field]]
 
         return cls(**data)

tests/test_patient_resource.py

@@ -84,3 +84,81 @@ def test_bulk_fhir_to_flat_patient():
     df.reset_index(inplace=True, drop=True)
     assert_frame_equal(pd.DataFrame(patient_ndjson_out), df)
     os.remove("multi_patient_output.parquet")
+
+
+PATIENT_EXT_DICT_INPUT = {
+    "id": "f001",
+    "active": True,
+    "extension": [
+        {"url": "Age", "valueQuantity": {"value": 25, "unit": "years"}},
+        {
+            "url": "birthSex",
+            "valueCodeableConcept": {
+                "coding": [
+                    {
+                        "system": "http://snomed.info/sct",
+                        "code": "248152002",
+                        "display": "Female (finding)",
+                    }
+                ]
+            },
+        },
+    ],
+    "name": [{"text": "Minnie Mouse"}],
+    "gender": "female",
+    "deceasedBoolean": False,
+}
+
+PATIENT_EXT_FLAT = {
+    "resourceType": "Patient",
+    "id": "f001",
+    "extension.Age.value": 25,
+    "extension.Age.unit": "years",
+    "extension.birthSex.code": "http://snomed.info/sct|248152002",
+    "extension.birthSex.text": "Female (finding)",
+    "gender": "female",
+    "deceasedBoolean": False,
+}
+
+PATIENT_EXT_DICT_OUT = {
+    "id": "f001",
+    "extension": [
+        {"url": "Age", "valueQuantity": {"value": 25, "unit": "years"}},
+        {
+            "url": "birthSex",
+            "valueCodeableConcept": {
+                "coding": [
+                    {
+                        "system": "http://snomed.info/sct",
+                        "code": "248152002",
+                        "display": "Female (finding)",
+                    }
+                ]
+            },
+        },
+    ],
+    "gender": "female",
+    "deceasedBoolean": False,
+}
+
+
+def test_patient_with_extensions_to_flat():
+    patient = Patient(**PATIENT_EXT_DICT_INPUT)
+
+    patient.to_flat("test_patient_ext.parquet")
+
+    assert_frame_equal(
+        pd.read_parquet("test_patient_ext.parquet"),
+        pd.DataFrame(PATIENT_EXT_FLAT, index=[0]),
+        check_like=True,  # ignore column order
+        check_dtype=False,
+    )
+    os.remove("test_patient_ext.parquet")
+
+
+def test_patient_with_extensions_from_flat():
+    patient = Patient(**PATIENT_EXT_DICT_OUT)
+
+    flat_patient = Patient.from_flat("tests/data/patient_ext_flat.parquet")
+
+    assert patient == flat_patient