Merge pull request #397 from icanbwell/safe-merge-enable

Enable safe merge

Author: changsu-bWell

Pipfile

@@ -39,7 +39,7 @@ bounded-pool-executor = ">=0.0.3"
 # fastjsonschema is needed for validating JSON
 fastjsonschema= ">=2.18.0"
 # helix.fhir.client.sdk is needed for interacting with FHIR servers
-"helix.fhir.client.sdk" = ">=4.1.11"
+"helix.fhir.client.sdk" = ">=4.1.28"
 # opensearch-py is needed for interacting with OpenSearch
 opensearch-py= { extras = ['async'], version = ">=2.6.0" }
 # pyathena is needed for interacting with Athena in AWS

Pipfile.lock

@@ -92,6 +92,7 @@ def __init__(
         drop_fields_from_json: Optional[List[str]] = None,
         partition_by_column_name: Optional[str] = None,
         enable_repartitioning: bool = True,
+        smart_merge: Optional[bool] = None,
     ):
         """
         Sends FHIR json stored in a folder to a FHIR server
@@ -274,6 +275,10 @@ def __init__(
         )
         self._setDefault(enable_repartitioning=enable_repartitioning)
 
+        self.smart_merge: Param[Optional[bool]] = Param(self, "smart_merge", "")
+
+        self._setDefault(smart_merge=None)
+
         kwargs = self._input_kwargs
         self.setParams(**kwargs)
 
@@ -317,6 +322,7 @@ def _transform(self, df: DataFrame) -> DataFrame:
         partition_by_column_name: Optional[str] = self.getOrDefault(
             self.partition_by_column_name
         )
+        smart_merge: Optional[bool] = self.getOrDefault(self.smart_merge)
 
         if not batch_size or batch_size == 0:
             batch_size = 30
@@ -487,6 +493,7 @@ def _transform(self, df: DataFrame) -> DataFrame:
                             validation_server_url=validation_server_url,
                             retry_count=retry_count,
                             exclude_status_codes_from_retry=exclude_status_codes_from_retry,
+                            smart_merge=smart_merge,
                         )
                     )
                     result_rows: List[Dict[str, Any]] = flatten(result_rows_list)
@@ -522,6 +529,7 @@ def _transform(self, df: DataFrame) -> DataFrame:
                             validation_server_url=validation_server_url,
                             retry_count=retry_count,
                             exclude_status_codes_from_retry=exclude_status_codes_from_retry,
+                            smart_merge=smart_merge,
                         )
                     )
                     rdd = (

spark_pipeline_framework/transformers/fhir_sender/v1/fhir_sender.py

@@ -33,6 +33,7 @@ def send_json_bundle_to_fhir(
     retry_count: Optional[int] = None,
     exclude_status_codes_from_retry: Optional[List[int]] = None,
     additional_request_headers: Optional[Dict[str, str]] = None,
+    smart_merge: Optional[bool] = None,
 ) -> Optional[FhirMergeResponse]:
     assert id_, f"{json_data_list!r}"
     fhir_client: FhirClient = get_fhir_client(
@@ -46,6 +47,8 @@ def send_json_bundle_to_fhir(
         auth_scopes=auth_scopes,
         log_level=log_level,
     )
+    if smart_merge is False:
+        fhir_client.smart_merge(False)
 
     fhir_client = fhir_client.resource(resource)
     if validation_server_url:

spark_pipeline_framework/transformers/fhir_sender/v1/fhir_sender_helpers.py

@@ -44,6 +44,7 @@ def send_partition_to_server(
         validation_server_url: Optional[str],
         retry_count: Optional[int],
         exclude_status_codes_from_retry: Optional[List[int]],
+        smart_merge: Optional[bool] = None,
     ) -> Generator[List[Dict[str, Any]], None, None]:
         """
         This function processes a partition
@@ -191,6 +192,7 @@ def send_partition_to_server(
                             log_level=log_level,
                             retry_count=retry_count,
                             exclude_status_codes_from_retry=exclude_status_codes_from_retry,
+                            smart_merge=smart_merge,
                         )
                         if result:
                             auth_access_token1 = result.access_token

spark_pipeline_framework/transformers/fhir_sender/v1/fhir_sender_processor.py

@@ -104,6 +104,7 @@ def __init__(
         max_chunk_size: int = 100,
         process_chunks_in_parallel: Optional[bool] = True,
         maximum_concurrent_tasks: int = 100,
+        smart_merge: Optional[bool] = None,
     ):
         """
         Sends FHIR json stored in a folder to a FHIR server
@@ -139,6 +140,7 @@ def __init__(
         :param max_chunk_size: (Optional) max chunk size
         :param process_chunks_in_parallel: (Optional) process chunks in parallel
         :param maximum_concurrent_tasks: (Optional) maximum concurrent tasks
+        :param smart_merge: (Optional) whether to use smart merge or not
         """
         super().__init__(
             name=name, parameters=parameters, progress_logger=progress_logger
@@ -307,6 +309,9 @@ def __init__(
         )
         self._setDefault(maximum_concurrent_tasks=maximum_concurrent_tasks)
 
+        self.smart_merge: Param[Optional[bool]] = Param(self, "smart_merge", "")
+        self._setDefault(smart_merge=None)
+
         kwargs = self._input_kwargs
         self.setParams(**kwargs)
 
@@ -394,6 +399,8 @@ async def _transform_async(self, df: DataFrame) -> DataFrame:
         )
         maximum_concurrent_tasks: int = self.getOrDefault(self.maximum_concurrent_tasks)
 
+        smart_merge: Optional[bool] = self.getOrDefault(self.smart_merge)
+
         if parameters and parameters.get("flow_name"):
             user_agent_value = (
                 f"{parameters['team_name']}:helix.pipelines:{parameters['flow_name']}".replace(
@@ -531,6 +538,7 @@ async def _transform_async(self, df: DataFrame) -> DataFrame:
                         process_chunks_in_parallel=process_chunks_in_parallel,
                         maximum_concurrent_tasks=maximum_concurrent_tasks,
                     ),
+                    smart_merge=smart_merge,
                 )
                 if run_synchronously:
                     rows_to_send: List[Dict[str, Any]] = [

spark_pipeline_framework/transformers/fhir_sender/v2/fhir_sender.py

@@ -36,6 +36,7 @@ async def send_json_bundle_to_fhir_async(
     exclude_status_codes_from_retry: Optional[List[int]] = None,
     additional_request_headers: Optional[Dict[str, str]] = None,
     batch_size: Optional[int] = None,
+    smart_merge: Optional[bool] = None,
 ) -> AsyncGenerator[FhirMergeResponse, None]:
     """
     Send a JSON bundle to FHIR server
@@ -54,6 +55,8 @@ async def send_json_bundle_to_fhir_async(
         log_level=log_level,
         auth_well_known_url=auth_well_known_url,
     )
+    if smart_merge is False:
+        fhir_client.smart_merge(False)
 
     fhir_client = fhir_client.resource(resource)
     if validation_server_url:

spark_pipeline_framework/transformers/fhir_sender/v2/fhir_sender_helpers_async.py

@@ -30,3 +30,4 @@ class FhirSenderParameters:
     retry_count: Optional[int]
     exclude_status_codes_from_retry: Optional[List[int]]
     pandas_udf_parameters: AsyncPandasUdfParameters
+    smart_merge: Optional[bool]

spark_pipeline_framework/transformers/fhir_sender/v2/fhir_sender_parameters.py

@@ -359,6 +359,7 @@ async def send_partition_to_server_async(
                                 log_level=parameters.log_level,
                                 retry_count=parameters.retry_count,
                                 exclude_status_codes_from_retry=parameters.exclude_status_codes_from_retry,
+                                smart_merge=parameters.smart_merge,
                             ):
                                 if result:
                                     auth_access_token1 = result.access_token
@@ -401,6 +402,7 @@ async def send_partition_to_server_async(
                             log_level=parameters.log_level,
                             retry_count=parameters.retry_count,
                             exclude_status_codes_from_retry=parameters.exclude_status_codes_from_retry,
+                            smart_merge=parameters.smart_merge,
                         ):
                             if result and result.request_id:
                                 request_id_list.append(result.request_id)

spark_pipeline_framework/transformers/fhir_sender/v2/fhir_sender_processor.py

@@ -73,6 +73,7 @@ def __init__(
         operation: Union[
             FhirSenderOperation, str
         ] = FhirSenderOperation.FHIR_OPERATION_MERGE.value,
+        smart_merge: Optional[bool] = None,
     ):
         """
         Runs the auto-mappers, saves the result to Athena db and then sends the results to fhir server
@@ -97,6 +98,7 @@ def __init__(
                         }
         :param enable_repartitioning: Enable repartitioning or not, default True
         :param operation: The API operation to perform, such as merge, put, delete etc
+        :param smart_merge: (Optional) Enable smart merge functionality
         """
         super().__init__(
             name=name, parameters=parameters, progress_logger=progress_logger
@@ -190,6 +192,9 @@ def __init__(
         )
         self._setDefault(operation=operation)
 
+        self.smart_merge: Param[Optional[bool]] = Param(self, "smart_merge", "")
+        self._setDefault(smart_merge=smart_merge)
+
         kwargs = self._input_kwargs
         self.setParams(**kwargs)
 
@@ -214,6 +219,7 @@ def _transform(self, df: DataFrame) -> DataFrame:
         )
         enable_repartitioning: bool = self.getOrDefault(self.enable_repartitioning)
         operation: Union[FhirSenderOperation, str] = self.getOrDefault(self.operation)
+        smart_merge: Optional[bool] = self.getSmartMerge()
 
         assert parameters
         progress_logger = self.getProgressLogger()
@@ -347,6 +353,7 @@ def _transform(self, df: DataFrame) -> DataFrame:
                             ),
                             enable_repartitioning=enable_repartitioning,
                             operation=operation,
+                            smart_merge=smart_merge,
                         ).transform(df)
         return df
 
@@ -409,3 +416,7 @@ def getSendToFhir(self) -> Optional[bool]:
     # noinspection PyPep8Naming,PyMissingOrEmptyDocstring
     def getAdditionalRequestHeaders(self) -> Optional[Dict[str, str]]:
         return self.getOrDefault(self.additional_request_headers)
+
+    # noinspection PyPep8Naming,PyMissingOrEmptyDocstring
+    def getSmartMerge(self) -> Optional[bool]:
+        return self.getOrDefault(self.smart_merge)

spark_pipeline_framework/transformers/send_automapper_to_fhir/v1/automapper_to_fhir_transformer.py

@@ -75,6 +75,7 @@ def __init__(
             FhirSenderOperation, str
         ] = FhirSenderOperation.FHIR_OPERATION_MERGE.value,
         log_level: Optional[str] = None,
+        smart_merge: Optional[bool] = None,
     ):
         """
         Runs the auto-mappers, saves the result to Athena db and then sends the results to fhir server
@@ -99,6 +100,8 @@ def __init__(
                         }
         :param enable_repartitioning: Enable repartitioning or not, default True
         :param operation: The API operation to perform, such as merge, put, delete etc
+        :param log_level: (Optional) Log level to use
+        :param smart_merge: (Optional) Enable smart merge functionality
         """
         super().__init__(
             name=name, parameters=parameters, progress_logger=progress_logger
@@ -195,6 +198,9 @@ def __init__(
         self.log_level: Param[str] = Param(self, "log_level", "")
         self._setDefault(log_level=log_level)
 
+        self.smart_merge: Param[Optional[bool]] = Param(self, "smart_merge", "")
+        self._setDefault(smart_merge=smart_merge)
+
         kwargs = self._input_kwargs
         self.setParams(**kwargs)
 
@@ -219,6 +225,7 @@ async def _transform_async(self, df: DataFrame) -> DataFrame:
         )
         enable_repartitioning: bool = self.getOrDefault(self.enable_repartitioning)
         operation: Union[FhirSenderOperation, str] = self.getOrDefault(self.operation)
+        smart_merge: Optional[bool] = self.getSmartMerge()
 
         assert parameters is not None
         progress_logger = self.getProgressLogger()
@@ -350,6 +357,7 @@ async def _transform_async(self, df: DataFrame) -> DataFrame:
                             ),
                             operation=operation,
                             log_level=log_level,
+                            smart_merge=smart_merge,
                         ).transform_async(df)
         return df
 
@@ -409,6 +417,10 @@ def getAuthScopes(self) -> Optional[List[str]]:
     def getSendToFhir(self) -> Optional[bool]:
         return self.getOrDefault(self.send_to_fhir)
 
-    # noinspection PyPep8Naming,PyMissingOrEmptyDocstring
+    # no inspection PyPep8Naming,PyMissingOrEmptyDocstring
     def getAdditionalRequestHeaders(self) -> Optional[Dict[str, str]]:
         return self.getOrDefault(self.additional_request_headers)
+
+    # noinspection PyPep8Naming,PyMissingOrEmptyDocstring
+    def getSmartMerge(self) -> Optional[bool]:
+        return self.getOrDefault(self.smart_merge)