previous response to param mapping (#2620)

* previous response to param mapping

* Add test that includes `previous_response_to_param_mapping`

* Small fixes for orchestration output

* Update tests with how we're using names/services

* Missed a test

containers/orchestration/app/custom_configs/seed-ecr-viewer-config.json

@@ -26,19 +26,27 @@
       "endpoint": "/stamp-condition-extensions"
     },
     {
-      "name": "message_parser_values",
+      "name": "metadata_values",
       "service": "message_parser",
       "endpoint": "/parse_message",
       "params": {
         "message_format": "fhir",
         "parsing_schema_name": "ecr_viewer_metadata.json",
         "credential_manager": "azure"
       }
+    },
+    {
+      "name:": "save_bundle",
+      "service": "save_bundle",
+      "url": "${ECR_VIEWER_URL}",
+      "endpoint": "/api/save-fhir-data",
+      "params": {
+        "saveSource": "postgres"
+      },
+      "previous_response_to_param_mapping": {
+        "metadata_values": "metadata",
+        "stamped_ecr": "fhirBundle"
+      }
     }
-  ],
-  "outputs": [
-    "message_parser_values",
-    "stamped_ecr"
-  ],
-  "default-response": false
+  ]
 }

containers/orchestration/app/handlers/request_builders/ecr_viewer.py

@@ -32,12 +32,17 @@ def build_save_fhir_data_body(
             "workflow_params": workflow_params,
         },
     ):
+        if workflow_params.get("fhirBundle"):
+            fhirBundle = workflow_params["fhirBundle"].json()["extended_bundle"]
+        else:
+            fhirBundle = input_msg
+
         request = {
-            "fhirBundle": input_msg,
+            "fhirBundle": fhirBundle,
             "saveSource": workflow_params.get("saveSource"),
         }
 
         if workflow_params.get("metadata") is not None:
-            request["metadata"] = workflow_params.get("metadata")
+            request["metadata"] = workflow_params["metadata"].json()["parsed_values"]
 
         return request

containers/orchestration/app/services.py

@@ -184,16 +184,20 @@ async def call_apis(
             service = step["service"]
             endpoint = step["endpoint"]
             endpoint_name = endpoint.split("/")[-1]
-            params = step.get("params", None)
+            params = step.get("params", {})
+            previous_response_to_param_mapping = step.get(
+                "previous_response_to_param_mapping", None
+            )
             call_span.add_event(
                 "formatting parameters for service " + service,
                 attributes={
                     "service": service,
                     "endpoint": endpoint,
                     "endpoint_name": endpoint_name,
-                    "config_params": [f"{k}: {v}" for k, v in params.items()]
-                    if params is not None
-                    else "",
+                    "config_params": _param_dict_to_str(params),
+                    "previous_response_to_param_mapping": _param_dict_to_str(
+                        previous_response_to_param_mapping
+                    ),
                 },
             )
 
@@ -208,6 +212,10 @@ async def call_apis(
                     "response_extraction_handler": response_func.__str__(),
                 },
             )
+
+            if previous_response_to_param_mapping:
+                for k, v in previous_response_to_param_mapping.items():
+                    params[v] = responses[k]
             request_body = request_body_func(current_message, input, params)
             call_span.add_event("posting to `service_url` " + service_url)
             response = post_request(service_url, request_body)
@@ -231,7 +239,7 @@ async def call_apis(
                     ),
                     attributes={"status_code": service_response.status_code},
                 )
-                error_detail = f"Service {service} failed with error {service_response.msg_content}"
+                error_detail = f"Service {service} failed with error {service_response.msg_content}, endpoint: {endpoint_name}"
                 call_span.set_status(StatusCode(2), error_detail)
                 raise HTTPException(
                     status_code=service_response.status_code, detail=error_detail
@@ -257,6 +265,11 @@ async def call_apis(
                     "updating input data with building block modifications"
                 )
                 current_message = service_response.msg_content
-            responses[service] = response
+            name = step.get("name", service)
+            responses[name] = response
         call_span.set_status(StatusCode(1))
         return (response, responses)
+
+
+def _param_dict_to_str(a_dict: dict):
+    return [f"{k}: {v}" for k, v in (a_dict or {}).items()]

containers/orchestration/app/utils.py

@@ -222,7 +222,7 @@ def _combine_response_bundles(
         if "name" in obj and obj["name"] in processing_config["outputs"]
     ]
     for step in config:
-        resp.append({step["name"]: responses[step["service"]].json()})
+        resp.append({step["name"]: responses[step["name"]].json()})
 
     if ("default-response" in processing_config) and (
         not processing_config["default-response"]

containers/orchestration/tests/integration/test_orchestration.py

@@ -176,6 +176,29 @@ def test_success_save_to_ecr_viewer(setup, clean_up_db):
         assert orchestration_response.status_code == 200
 
 
+@pytest.mark.integration
+def test_previous_response_mapping_for_ecr_viewer(setup, clean_up_db):
+    """
+    Full orchestration test of a zip file containing both an eICR and the
+    associated RR data, using the `previous_response_to_param_mapping` in the config
+    """
+    with open(
+        Path(__file__).parent.parent / "assets" / "test_zip.zip",
+        "rb",
+    ) as file:
+        form_data = {
+            "message_type": "ecr",
+            "data_type": "zip",
+            "config_file_name": "seed-ecr-viewer-config.json",
+        }
+        files = {"upload_file": ("file.zip", file)}
+        orchestration_response = httpx.post(
+            PROCESS_ZIP_ENDPOINT, data=form_data, files=files, timeout=60
+        )
+
+        assert orchestration_response.status_code == 200
+
+
 @pytest.mark.integration
 def test_process_message_fhir(setup):
     """

containers/orchestration/tests/test_utils.py

@@ -103,8 +103,8 @@ def test_combine_response_bundles_with_outputs():
     combined = _combine_response_bundles(
         mock_response,
         {
-            "foobar": mock_response_2,
-            "bizboo": mock_response,
+            "foo": mock_response_2,
+            "biz": mock_response,
             "notIncluded": mock_response_3,
         },
         config,
@@ -129,8 +129,8 @@ def test_combine_response_bundles_without_outputs():
     combined = _combine_response_bundles(
         mock_response,
         {
-            "foobar": mock_response_2,
-            "bizboo": mock_response,
+            "foo": mock_response_2,
+            "biz": mock_response,
             "notIncluded": mock_response_3,
         },
         config_2,
@@ -151,8 +151,8 @@ def test_combine_response_bundles_with_default_response_off():
     combined = _combine_response_bundles(
         mock_response,
         {
-            "foobar": mock_response_2,
-            "bizboo": mock_response,
+            "foo": mock_response_2,
+            "biz": mock_response,
             "notIncluded": mock_response_3,
         },
         config,