Skip to content

Commit

Permalink
Fix write_csv test, update requested changes
Browse files Browse the repository at this point in the history
  • Loading branch information
@TraciebelWairimu
TraciebelWairimu committed Mar 1, 2024
1 parent 63fd41a commit 75a7bd7
Show file tree
Hide file tree
Showing 3 changed files with 79 additions and 77 deletions.
2 changes: 1 addition & 1 deletion importer/README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,6 @@ The coverage report `coverage.html` will be at the working directory
- `export_resources` can either be True or False, checks if it is True and exports the resources
- The `parameter` is used as a filter for the resources. The set default parameter is "_lastUpdated", other examples include, "name"
- The `value` is where you pass the actual parameter value to filter the resources. The set default value is "gt2023-01-01", other examples include, "Good Health Clinic 1"
- The `limit` is the number of resources exported at a time
- The `limit` is the number of resources exported at a time. The set default value is '1000'
- Specify the `resource_type` you want to export, different resource_types are exported to different csv_files
- The csv_file containing the exported resources is labelled using the current time, to know when the resources were exported for example, csv/2024-02-21-12-21-export_Location.csv
142 changes: 75 additions & 67 deletions importer/main.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -886,7 +886,10 @@ def write_csv(data, resource_type, fieldnames):
with open(csv_file, 'w', newline='') as file:
csv_writer = csv.writer(file)
csv_writer.writerow(fieldnames)
csv_writer.writerows(data)
with click.progressbar(data, label='Progress:: Writing csv') as write_csv_progress:
for row in write_csv_progress:
csv_writer.writerow(row)
return csv_file


def get_base_url():
Expand All @@ -905,73 +908,78 @@ def export_resources_to_csv(resource_type, parameter, value, limit):
if response[1] == 200:
resources = json.loads(response[0])
data = []
if "entry" in resources:
if resource_type == "Location":
elements = ["name", "status", "method", "id", "identifier", "parentName", "parentID", "type",
"typeCode",
"physicalType", "physicalTypeCode"]
elif resource_type == "Organization":
elements = ["name", "active", "method", "id", "identifier", "alias"]
elif resource_type == "CareTeam":
elements = ["name", "status", "method", "id", "identifier", "organizations", "participants"]
try:
if resources["entry"]:
if resource_type == "Location":
elements = ["name", "status", "method", "id", "identifier", "parentName", "parentID", "type",
"typeCode",
"physicalType", "physicalTypeCode"]
elif resource_type == "Organization":
elements = ["name", "active", "method", "id", "identifier", "alias"]
elif resource_type == "CareTeam":
elements = ["name", "status", "method", "id", "identifier", "organizations", "participants"]
else:
elements = []
with click.progressbar(resources["entry"],
label='Progress:: Extracting resource') as extract_resources_progress:
for x in extract_resources_progress:
rl = []
orgs_list = []
participants_list = []
for element in elements:
try:
if element == "method":
value = "update"
elif element == "active":
value = x["resource"]["active"]
elif element == "identifier":
value = x["resource"]["identifier"][0]["value"]
elif element == "organizations":
organizations = x["resource"]["managingOrganization"]
for index, value in enumerate(organizations):
reference = x["resource"]["managingOrganization"][index]["reference"]
new_reference = reference.split("/", 1)[1]
display = x["resource"]["managingOrganization"][index]["display"]
organization = ":".join([new_reference, display])
orgs_list.append(organization)
string = "|".join(map(str, orgs_list))
value = string
elif element == "participants":
participants = x["resource"]["participant"]
for index, value in enumerate(participants):
reference = x["resource"]["participant"][index]["member"]["reference"]
new_reference = reference.split("/", 1)[1]
display = x["resource"]["participant"][index]["member"]["display"]
participant = ":".join([new_reference, display])
participants_list.append(participant)
string = "|".join(map(str, participants_list))
value = string
elif element == "parentName":
value = x["resource"]["partOf"]["display"]
elif element == "parentID":
reference = x["resource"]["partOf"]["reference"]
value = reference.split("/", 1)[1]
elif element == "type":
value = x["resource"]["type"][0]["coding"][0]["display"]
elif element == "typeCode":
value = x["resource"]["type"][0]["coding"][0]["code"]
elif element == "physicalType":
value = x["resource"]["physicalType"]["coding"][0]["display"]
elif element == "physicalTypeCode":
value = x["resource"]["physicalType"]["coding"][0]["code"]
elif element == "alias":
value = x["resource"]["alias"][0]
else:
value = x["resource"][element]
except KeyError:
value = ""
rl.append(value)
data.append(rl)
write_csv(data, resource_type, elements)
logging.info("Successfully written to csv")
else:
elements = []
for x in resources["entry"]:
rl = []
orgs_list = []
participants_list = []
for element in elements:
try:
if element == "method":
value = "update"
elif element == "active":
value = x["resource"]["active"]
elif element == "identifier":
value = x["resource"]["identifier"][0]["value"]
elif element == "organizations":
organizations = x["resource"]["managingOrganization"]
for index, value in enumerate(organizations):
reference = x["resource"]["managingOrganization"][index]["reference"]
new_reference = reference.split("/", 1)[1]
display = x["resource"]["managingOrganization"][index]["display"]
organization = ":".join([new_reference, display])
orgs_list.append(organization)
string = "|".join(map(str, orgs_list))
value = string
elif element == "participants":
participants = x["resource"]["participant"]
for index, value in enumerate(participants):
reference = x["resource"]["participant"][index]["member"]["reference"]
new_reference = reference.split("/", 1)[1]
display = x["resource"]["participant"][index]["member"]["display"]
participant = ":".join([new_reference, display])
participants_list.append(participant)
string = "|".join(map(str, participants_list))
value = string
elif element == "parentName":
value = x["resource"]["partOf"]["display"]
elif element == "parentID":
reference = x["resource"]["partOf"]["reference"]
value = reference.split("/", 1)[1]
elif element == "type":
value = x["resource"]["type"][0]["coding"][0]["display"]
elif element == "typeCode":
value = x["resource"]["type"][0]["coding"][0]["code"]
elif element == "physicalType":
value = x["resource"]["physicalType"]["coding"][0]["display"]
elif element == "physicalTypeCode":
value = x["resource"]["physicalType"]["coding"][0]["code"]
elif element == "alias":
value = x["resource"]["alias"][0]
else:
value = x["resource"][element]
except KeyError:
value = ""
rl.append(value)
data.append(rl)
write_csv(data, resource_type, elements)
logging.info("Successfully written to csv")
else:
logging.info("No entry found")
except KeyError:
logging.info("No Resources Found")
else:
logging.error(f"Failed to retrieve resource. Status code: {response[1]} response: {response[0]}")
Expand Down
12 changes: 3 additions & 9 deletions importer/test_main.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import json
import unittest
from datetime import datetime
from jsonschema import validate
from mock import patch
from main import (
Expand Down Expand Up @@ -42,14 +41,9 @@ def test_write_csv(self):
]
self.test_resource_type = "test_organization"
self.test_fieldnames = ["name", "active", "method", "id", "identifier", "alias"]
write_csv(self.test_data, self.test_resource_type, self.test_fieldnames)
self.assertIsInstance(self.test_data, list)
self.assertEqual(len(self.test_data), 2)
current_time = datetime.now().strftime("%Y-%m-%d-%H-%M")
expected_csv_file_path = (
f"csv/exports/{current_time}-export_{self.test_resource_type}.csv"
)
self.assertTrue(expected_csv_file_path, "CSV file created in expected location")
csv_file = write_csv(self.test_data, self.test_resource_type, self.test_fieldnames)
csv_content = read_csv(csv_file)
self.assertEqual(csv_content, self.test_data)

@patch("main.get_resource")
def test_build_payload_organizations(self, mock_get_resource):
Expand Down

0 comments on commit 75a7bd7

Please sign in to comment.