PRMP-640 (#11)

* [PRMP-640] - changes to enrichment lambda

lambda/event-enrichment/event_enrichment_main.py

@@ -1,10 +1,13 @@
 import json
 import os
+from datetime import date, timedelta, datetime
 from typing import Optional
 
 import boto3
 import urllib3
 
+from services.models.ods_models import PracticeOds, IcbOds
+
 ODS_PORTAL_URL = "https://directory.spineservices.nhs.uk/ORD/2-0-0/organisations/"
 ICB_ROLE_ID = "RO98"
 EMPTY_ORGANISATION = {"Name": None}
@@ -65,55 +68,133 @@ def _enrich_events(sqs_messages: dict) -> list:
                 f"Skipping enrichment for degrades event with eventId: {event['eventId']}."
             )
             continue
-
         # set requesting practice info
-        requesting_practice_organisation = _fetch_organisation(
-            event["requestingPracticeOdsCode"]
+        event.update(
+            _requesting_practice_info(
+                ods_code=event["requestingPracticeOdsCode"],
+                practice_name_key="requestingPracticeName",
+                icb_name_key="requestingPracticeIcbName",
+                icb_ods_code_key="requestingPracticeIcbOdsCode",
+                supplier_key="requestingSupplierName",
+            )
         )
-        event["requestingPracticeName"] = requesting_practice_organisation["Name"]
-        event["requestingPracticeIcbOdsCode"] = _find_icb_ods_code(
-            requesting_practice_organisation
+        # set sending practice info
+        event.update(
+            _requesting_practice_info(
+                ods_code=event["sendingPracticeOdsCode"],
+                practice_name_key="sendingPracticeName",
+                icb_name_key="sendingPracticeIcbName",
+                icb_ods_code_key="sendingPracticeIcbOdsCode",
+                supplier_key="sendingSupplierName",
+            )
         )
-        event["requestingPracticeIcbName"] = _fetch_organisation(
-            event["requestingPracticeIcbOdsCode"]
-        )["Name"]
 
-        # set sending practice info
-        sending_practice_organisation = _fetch_organisation(
-            event["sendingPracticeOdsCode"]
+        # temporary fix for EMIS wrong reportingSystemSupplier data
+        reporting_system_supplier = event["reportingSystemSupplier"]
+        if reporting_system_supplier.isnumeric():
+            print(
+                f"TEMP FIX. Reporting system supplier received: {reporting_system_supplier}. Changed to 'EMIS'."
+            )
+            event["reportingSystemSupplier"] = "EMIS"
+
+    return events
+
+
+def _requesting_practice_info(
+    ods_code: str, practice_name_key: str, icb_name_key: str, icb_ods_code_key: str, supplier_key: str
+) -> dict:
+    enrichment_info = {}
+    print("requesting data for: " + ods_code)
+    gp_dynamo_item = arrange_gp_data_from_dynamo(ods_code) or get_gp_data_from_api(
+        ods_code
+    )
+    if gp_dynamo_item:
+        enrichment_info.update(
+            {
+                practice_name_key: gp_dynamo_item.practice_name,
+                icb_ods_code_key: gp_dynamo_item.icb_ods_code,
+            }
         )
-        event["sendingPracticeName"] = sending_practice_organisation["Name"]
-        event["sendingPracticeIcbOdsCode"] = _find_icb_ods_code(
-            sending_practice_organisation
+        enrichment_info[supplier_key] = (
+            get_supplier_data(ods_code, gp_dynamo_item) or "UNKNOWN"
         )
-        event["sendingPracticeIcbName"] = _fetch_organisation(
-            event["sendingPracticeIcbOdsCode"]
-        )["Name"]
-
-        # set requesting supplier info
-        requesting_supplier_name = get_supplier_name(event["requestingPracticeOdsCode"])
-        event["requestingSupplierName"] = (
-            requesting_supplier_name
-            if requesting_supplier_name is not None
-            else "UNKNOWN"
+        enrichment_info[icb_name_key] = (
+            get_icb_name(gp_dynamo_item.icb_ods_code) or "UNKNOWN"
         )
+    else:
+        enrichment_info[supplier_key] = "UNKNOWN"
+    return enrichment_info
 
-        # set sending supplier info
-        sending_supplier_name = get_supplier_name(event["sendingPracticeOdsCode"])
-        event["sendingSupplierName"] = (
-            sending_supplier_name 
-            if sending_supplier_name is not None 
-            else "UNKNOWN"
-        )
 
-        # temporary fix for EMIS wrong reportingSystemSupplier data        
-        reporting_system_supplier = event["reportingSystemSupplier"]
-        if reporting_system_supplier.isnumeric():
-            print(f"TEMP FIX. Reporting system supplier received: {reporting_system_supplier}. Changed to 'EMIS'.")
-            event["reportingSystemSupplier"] = "EMIS"            
+def arrange_gp_data_from_dynamo(ods_code: str):
+    try:
+        gp_dynamo_data = get_gp_data_from_dynamo_request(ods_code)
+        print("Successfully query dynamo for GP data")
+        return gp_dynamo_data
+    except PracticeOds.DoesNotExist:
+        print("Failed to find GP data in dynamo table")
+        return None
 
 
-    return events
+def get_icb_name(ods_code: str):
+    if ods_code is None:
+        return None
+    else:
+        return get_icb_name_from_dynamo(ods_code) or get_icb_name_from_api(ods_code)
+
+
+def get_icb_name_from_dynamo(ods_code: str):
+    try:
+        icb_dynamo_item = get_icb_data_from_dynamo_request(ods_code)
+        print("Successfully query dynamo for ICB data")
+        return icb_dynamo_item.icb_name
+    except IcbOds.DoesNotExist:
+        print("Failed to find ICB data in dynamo table")
+        return None
+
+
+def get_gp_data_from_api(ods_code: str):
+    requesting_practice_organisation = _fetch_organisation(ods_code)
+
+    practice_name = requesting_practice_organisation["Name"]
+    if practice_name is None:
+        return None
+    icb_ods_code = _find_icb_ods_code(requesting_practice_organisation)
+    gp_api_item = PracticeOds(ods_code, practice_name, icb_ods_code=icb_ods_code)
+    gp_api_item.save()
+    return gp_api_item
+
+
+def get_icb_name_from_api(ods_code: str):
+    icb_name = _fetch_organisation(ods_code)["Name"]
+    icb_api_item = IcbOds(ods_code, icb_name)
+    icb_api_item.save()
+    return icb_name
+
+
+def get_supplier_data(ods_code: str, gp_dynamo_item: PracticeOds):
+    date_today = date.today()
+    date_one_month_ago = date_today - timedelta(days=30)
+    supplier_last_update_date = (
+        gp_dynamo_item.supplier_last_updated.date()
+        if gp_dynamo_item.supplier_last_updated
+        else None
+    )
+    is_out_of_date = (
+        supplier_last_update_date < date_one_month_ago
+        if supplier_last_update_date
+        else True
+    )
+    if not gp_dynamo_item.supplier_name and is_out_of_date:
+        requesting_supplier_name = get_supplier_name_from_sds_api(ods_code)
+        gp_dynamo_item.supplier_name = requesting_supplier_name
+        gp_dynamo_item.update(
+            actions=[
+                PracticeOds.supplier_name.set(requesting_supplier_name),
+                PracticeOds.supplier_last_updated.set(datetime.now()),
+            ]
+        )
+    return gp_dynamo_item.supplier_name
 
 
 def _find_icb_ods_code(practice_organisation: dict) -> Optional[str]:
@@ -260,9 +341,9 @@ def _find_supplier_ods_codes_from_supplier_details(supplier_details: dict) -> li
     return supplier_ods_codes
 
 
-def get_supplier_name(practice_ods_code: str) -> Optional[str]:
+def get_supplier_name_from_sds_api(practice_ods_code: str) -> Optional[str]:
     """uses the SDS FHIR API to get the system supplier from an ODS code"""
-
+    print("Requesting supplier info from SDS")
     if not practice_ods_code or practice_ods_code.isspace():
         return None
 
@@ -278,21 +359,27 @@ def get_supplier_name(practice_ods_code: str) -> Optional[str]:
     }
 
     supplier_name = None
-    
+
     for supplier_ods_code in supplier_ods_codes:
         try:
             supplier_name = supplier_name_mapping[supplier_ods_code]
             if supplier_name is not None:
                 break
         except KeyError:
             continue
-    
-    if supplier_name is None:        
+
+    if supplier_name is None:
         print(
             f"Unable to map supplier ODS code(s) found from SDS FHI API: {str(supplier_ods_codes)}"
             + " to a known supplier name. Practice ODS code from event: {practice_ods_code}."
-        )    
+        )
 
     return supplier_name
-
-
+
+
+def get_gp_data_from_dynamo_request(ods_code: str):
+    return PracticeOds.get(ods_code)
+
+
+def get_icb_data_from_dynamo_request(ods_code: str):
+    return IcbOds.get(ods_code)

lambda/event-enrichment/requirements.txt

@@ -0,0 +1 @@
+pynamodb==6.0.1

lambda/event-enrichment/test_event_enrichment_main.py

@@ -18,7 +18,7 @@
     _find_supplier_ods_codes_from_supplier_details,
     _has_supplier_ods_code,
     UnableToFetchSupplierDetailsFromSDSFHIRException,
-    get_supplier_name,
+    get_supplier_name_from_sds_api,
     UnableToMapSupplierOdsCodeToSupplierNameException,
 )
 
@@ -626,7 +626,7 @@ def test_returns_supplier_name_given_a_practice_ods_code(
             {"Parameter": {"Value": "some_url.net?"}},
         ]
 
-        supplier_name = get_supplier_name("test_supplier_ods_code")
+        supplier_name = get_supplier_name_from_sds_api("test_supplier_ods_code")
 
         expected_supplier_name = "EMIS"
         assert supplier_name == expected_supplier_name
@@ -646,7 +646,7 @@ def test_supplier_name_returns_none_when_supplier_ods_code_is_not_found_from_sds
             {"Parameter": {"Value": "some_url.net?"}},
         ]
 
-        supplier_name = get_supplier_name("PRACTICE_ODS_123")
+        supplier_name = get_supplier_name_from_sds_api("PRACTICE_ODS_123")
 
         assert supplier_name is None
 
@@ -665,6 +665,6 @@ def test_supplier_name_returns_none_when_supplier_ods_code_is_none(
             {"Parameter": {"Value": "some_url.net?"}},
         ]
 
-        supplier_name = get_supplier_name(None)
+        supplier_name = get_supplier_name_from_sds_api(None)
 
         assert supplier_name is None

services/models/ods_models.py

@@ -0,0 +1,22 @@
+import os
+
+from pynamodb.attributes import UnicodeAttribute, UTCDateTimeAttribute
+from pynamodb.models import Model
+
+class PracticeOds(Model):
+    class Meta:
+        table_name = os.getenv("GP_ODS_DYNAMO_TABLE_NAME")
+
+    practice_ods_code = UnicodeAttribute(hash_key=True, attr_name='PracticeOdsCode')
+    practice_name = UnicodeAttribute(attr_name='PracticeName')
+    icb_ods_code = UnicodeAttribute(null=True, attr_name='IcbOdsCode')
+    supplier_name = UnicodeAttribute(null=True, attr_name='SupplierName')
+    supplier_last_updated = UTCDateTimeAttribute(null=True, attr_name='SupplierLastUpdated')
+
+class IcbOds(Model):
+    class Meta:
+        table_name = os.getenv("ICB_ODS_DYNAMO_TABLE_NAME")
+
+    icb_ods_code = UnicodeAttribute(hash_key=True, attr_name='IcbOdsCode')
+    icb_name = UnicodeAttribute(attr_name='IcbName')
+