From 1911eb1d5d412b7302b5eb844cd141cfb6c12312 Mon Sep 17 00:00:00 2001 From: Wambere Date: Fri, 26 Jul 2024 15:00:47 +0300 Subject: [PATCH] Add config option to define own location-type coding system url (#239) --- importer/README.md | 1 + importer/json_payloads/locations_payload.json | 2 +- importer/main.py | 19 ++++++++++---- importer/test_main.py | 25 ++++++++++++++++--- 4 files changed, 38 insertions(+), 9 deletions(-) diff --git a/importer/README.md b/importer/README.md index 7680e2de..9a89ffa4 100644 --- a/importer/README.md +++ b/importer/README.md @@ -91,6 +91,7 @@ The coverage report `coverage.html` will be at the working directory - The seventh and eighth columns are the location's type and typeCode, respectively - The ninth column is the administrative level, that shows the hierarchical level of the location. Root location would have a `level 0` and all child locations will have a level `parent_admin_level + 1` - The tenth and eleventh columns are the location's physicalType and physicalTypeCode, respectively +- You can pass in `--location_type_coding_system` to define your own location type coding system url (not required) ### 2. Create users in bulk - Run `python3 main.py --csv_file csv/users.csv --resource_type users --log_level info` diff --git a/importer/json_payloads/locations_payload.json b/importer/json_payloads/locations_payload.json index 553f5d3e..5506ca03 100644 --- a/importer/json_payloads/locations_payload.json +++ b/importer/json_payloads/locations_payload.json @@ -23,7 +23,7 @@ { "coding": [ { - "system": "http://terminology.hl7.org/CodeSystem/location-type", + "system": "$t_system", "code": "$t_code", "display": "$t_display" } diff --git a/importer/main.py b/importer/main.py index 8a00a437..efe6cae9 100644 --- a/importer/main.py +++ b/importer/main.py @@ -233,6 +233,8 @@ def identify_coding_object_index(array, current_system): list_of_systems = value["coding"][0]["system"] if current_system in list_of_systems: return index + else: + return -1 def check_parent_admin_level(locationParentId): @@ -256,7 +258,7 @@ def check_parent_admin_level(locationParentId): # custom extras for locations -def location_extras(resource, payload_string): +def location_extras(resource, payload_string, location_coding_system): try: ( locationName, @@ -302,7 +304,8 @@ def location_extras(resource, payload_string): payload_string = json.dumps(obj, indent=4) try: - if locationType and locationType != "type": + payload_string = payload_string.replace("$t_system", location_coding_system) + if len(locationType.strip()) > 0 and locationType != "type": payload_string = payload_string.replace("$t_display", locationType) if locationTypeCode and locationTypeCode != "typeCode": payload_string = payload_string.replace("$t_code", locationTypeCode) @@ -975,7 +978,7 @@ def check_for_nulls(resource: list) -> list: # This function builds a json payload # which is posted to the api to create resources -def build_payload(resource_type, resources, resource_payload_file, created_resources=None): +def build_payload(resource_type, resources, resource_payload_file, created_resources=None, location_coding_system=None): logging.info("Building request payload") initial_string = """{"resourceType": "Bundle","type": "transaction","entry": [ """ final_string = group_type = " " @@ -1037,7 +1040,7 @@ def build_payload(resource_type, resources, resource_payload_file, created_resou if resource_type == "organizations": ps = organization_extras(resource, ps) elif resource_type == "locations": - ps = location_extras(resource, ps) + ps = location_extras(resource, ps, location_coding_system) elif resource_type == "careTeams": ps = care_team_extras(resource, ps, "orgs & users") elif resource_type == "Group": @@ -1808,6 +1811,10 @@ def filter(self, record): required=False, default="DIRECT", ) +@click.option( + "--location_type_coding_system", + required=False, + default="http://terminology.hl7.org/CodeSystem/location-type") def main( csv_file, json_file, @@ -1829,6 +1836,7 @@ def main( resources_count, list_resource_id, sync, + location_type_coding_system, ): if log_level == "DEBUG": logging.basicConfig( @@ -1895,7 +1903,8 @@ def main( elif resource_type == "locations": logging.info("Processing locations") json_payload = build_payload( - "locations", resource_list, "json_payloads/locations_payload.json" + "locations", resource_list, "json_payloads/locations_payload.json", None, + location_type_coding_system ) final_response = handle_request("POST", json_payload, fhir_base_url) logging.info("Processing complete!") diff --git a/importer/test_main.py b/importer/test_main.py index 1e35fa11..f0fec1a1 100644 --- a/importer/test_main.py +++ b/importer/test_main.py @@ -150,7 +150,8 @@ def test_build_payload_locations( csv_file = "csv/locations/locations_full.csv" resource_list = read_csv(csv_file) payload = build_payload( - "locations", resource_list, "json_payloads/locations_payload.json" + "locations", resource_list, "json_payloads/locations_payload.json", + None, "http://terminology.hl7.org/CodeSystem/location-type" ) payload_obj = json.loads(payload) self.assertIsInstance(payload_obj, dict) @@ -239,7 +240,8 @@ def test_build_payload_locations( csv_file = "csv/locations/locations_min.csv" resource_list = read_csv(csv_file) payload = build_payload( - "locations", resource_list, "json_payloads/locations_payload.json" + "locations", resource_list, "json_payloads/locations_payload.json", + None, "http://terminology.hl7.org/CodeSystem/location-type" ) payload_obj = json.loads(payload) self.assertIsInstance(payload_obj, dict) @@ -626,7 +628,8 @@ def test_uuid_generated_for_locations_is_unique_and_repeatable(self): ] payload = build_payload( - "locations", resources, "json_payloads/locations_payload.json" + "locations", resources, "json_payloads/locations_payload.json", + None, "http://terminology.hl7.org/CodeSystem/location-type" ) payload_obj = json.loads(payload) location1 = payload_obj["entry"][0]["resource"]["id"] @@ -1535,6 +1538,22 @@ def test_build_resource_type_map(self): self.assertIsInstance(mapping, dict) self.assertEqual(mapping, mapped_resources) + @patch("main.check_parent_admin_level") + @patch("main.get_resource") + def test_define_own_location_type_coding_system_url(self, mock_get_resource, mock_check_parent_admin_level): + mock_get_resource.return_value = "1" + mock_check_parent_admin_level.return_value = "3" + test_system_code = "http://terminology.hl7.org/CodeSystem/test_location-type" + + csv_file = "csv/locations/locations_full.csv" + resource_list = read_csv(csv_file) + payload = build_payload( + "locations", resource_list, "json_payloads/locations_payload.json", + None, test_system_code + ) + payload_obj = json.loads(payload) + self.assertEqual(payload_obj["entry"][0]["resource"]["type"][0]["coding"][0]["system"], test_system_code) + if __name__ == "__main__": unittest.main()