Merge pull request #130 from icanbwell/TG-DFP-2882-main

DFP-2882 Added support to pass regex for reference

spark_auto_mapper_fhir/fhir_types/fhir_reference.py

@@ -19,15 +19,35 @@ def __init__(
         resource: str,
         column: Union[AutoMapperDataTypeColumn, AutoMapperTextLikeBase],
         use_long_id: Optional[bool] = False,
+        reference_pattern: str = "",
     ):
+        """
+        :param resource: (str) The resource for which we need to create a reference.
+        :param column: (obj) The column used to create the ID.
+        :param use_long_id: (bool) Indicates whether to use a long ID. If set to False, it will limit the ID to
+         63 characters; otherwise, it will be 1024*1024 characters.
+        :param reference_pattern: (str) The reference pattern to be applied to the column value. If passed, it will use
+        the specified reference pattern; otherwise, the default will be used.
+        """
         super().__init__()
 
         assert resource
         assert "/" not in resource
         self.resource: str = resource
         self.use_long_id = use_long_id
+        # Default reference pattern ignores the pipe character so that if our reference has value like
+        # "id|sourceAssigningAuthority" the pipe character does not get replaced. However, there are cases where we may
+        # need to replace the pipe character or specify a custom regex to ensure that FHIRReference matches FHIRId.
+        # For example, insurance reference may contain pipe characters and while creating ID pipe is removed so we
+        # should also remove it while creating reference so that FHIRId matches FHIRReference.
+        reference_pattern = (
+            r"[^A-Za-z0-9\|\-\.]" if not reference_pattern else reference_pattern
+        )
         self.column: Union[AutoMapperDataTypeColumn, AutoMapperTextLikeBase] = FhirId(
-            column, is_reference=True, use_long_id=self.use_long_id
+            column,
+            is_reference=True,
+            use_long_id=self.use_long_id,
+            reference_pattern=reference_pattern,
         )
 
     def get_column_spec(

spark_auto_mapper_fhir/fhir_types/id.py

@@ -16,12 +16,21 @@ def __init__(
         column: Union[AutoMapperDataTypeColumn, AutoMapperTextLikeBase],
         is_reference: Optional[bool] = False,
         use_long_id: Optional[bool] = False,
+        reference_pattern: str = "",
     ):
+        """
+        :param column: (obj) The column used to create the ID.
+        :param is_reference: (bool) Indicates if the ID is being created for a reference or not.
+        :param use_long_id: (bool) Indicates whether to use a long ID. If set to false, it will limit the ID to
+         63 characters; otherwise, it will be 1024*1024 characters.
+        :param reference_pattern: (str) The reference pattern to be applied to the column value.
+        """
         super().__init__()
 
         self.column: Union[AutoMapperDataTypeColumn, AutoMapperTextLikeBase] = column
         self.is_reference = is_reference
         self.use_long_id = use_long_id
+        self.reference_pattern = reference_pattern
 
     def get_column_spec(
         self,
@@ -37,7 +46,7 @@ def get_column_spec(
                     current_column=current_column,
                     parent_columns=parent_columns,
                 ),
-                pattern=r"[^A-Za-z0-9\|\-\.]"
+                pattern=self.reference_pattern
                 if self.is_reference
                 else r"[^A-Za-z0-9\-\.]",
                 replacement="-",

tests/fhir_types/fhir_reference/test_automapper_fhir_reference.py

@@ -9,7 +9,15 @@
 from spark_auto_mapper_fhir.complex_types.reference import Reference
 from spark_auto_mapper_fhir.fhir_types.fhir_reference import FhirReference
 from spark_auto_mapper_fhir.fhir_types.id import FhirId
+from spark_auto_mapper_fhir.fhir_types.list import FhirList
 from spark_auto_mapper_fhir.resources.patient import Patient
+from spark_auto_mapper_fhir.resources.practitioner_role import PractitionerRole
+from spark_auto_mapper_fhir.extensions.custom.insurance_plan import (
+    InsurancePlanExtension,
+)
+from spark_auto_mapper_fhir.extensions.custom.insurance_plan_item import (
+    InsurancePlanItemExtension,
+)
 
 
 def test_auto_mapper_fhir_reference_short(spark_session: SparkSession) -> None:
@@ -150,3 +158,71 @@ def test_auto_mapper_fhir_reference_long(spark_session: SparkSession) -> None:
         .collect()[0][0]
         == "Organization/1F238EABDF31F1FFEDE1D316B5887.922A7B6A6EB4D9C82C9E7FFC748E8-7C28346B3999CB6F|test"
     )
+
+
+def test_auto_mapper_fhir_reference_reference_pattern(
+    spark_session: SparkSession,
+) -> None:
+    """
+    Test reference_pattern is used when pattern for a reference is passed.
+    """
+    # Arrange
+    spark_session.createDataFrame(
+        [
+            ("AETNA", "First Best Plan |  For Member"),
+            ("UHC", "Second Best Plan   For Member"),
+        ],
+        ["insurance_name", "insurance_plan"],
+    ).createOrReplaceTempView("insurance")
+
+    source_df: DataFrame = spark_session.table("insurance")
+    source_df.createOrReplaceTempView("insurance_plan")
+
+    # Act
+    mapper = AutoMapper(
+        view="insurance_plan", source_view="insurance", keys=["insurance_name"]
+    ).complex(
+        PractitionerRole(
+            id_=FhirId(
+                A.concat(A.column("insurance_name"), "-", A.column("insurance_plan"))
+            ),
+            extension=FhirList(
+                [
+                    InsurancePlanExtension(
+                        url=InsurancePlanExtension.codeset,
+                        extension=FhirList(
+                            [
+                                InsurancePlanItemExtension(
+                                    url=InsurancePlanItemExtension.codeset,
+                                    valueReference=Reference(
+                                        reference=FhirReference(
+                                            "InsurancePlan",
+                                            A.column("insurance_plan"),
+                                            reference_pattern=r"[^A-Za-z0-9\-\.]",
+                                        )
+                                    ),
+                                )
+                            ]
+                        ),
+                    )
+                ]
+            ),
+        )
+    )
+
+    # Assert
+    assert isinstance(mapper, AutoMapper)
+    result_df: DataFrame = mapper.transform(df=source_df)
+
+    assert (
+        result_df.where(result_df["id"] == "AETNA-First-Best-Plan----For-Member")
+        .selectExpr("extension[0].extension[0].valueReference.reference")
+        .collect()[0][0]
+        == "InsurancePlan/First-Best-Plan----For-Member"
+    )
+    assert (
+        result_df.where(result_df["id"] == "UHC-Second-Best-Plan---For-Member")
+        .selectExpr("extension[0].extension[0].valueReference.reference")
+        .collect()[0][0]
+        == "InsurancePlan/Second-Best-Plan---For-Member"
+    )