Adding insert for NPH Participant durring PPSC Opt In Sync (#4067)

* add insert got nph participant table

* fix tests

Author: dtharpeuno

rdr_service/dao/ppsc_dao.py

@@ -7,6 +7,7 @@
 from rdr_service.dao.base_dao import BaseDao, UpsertableDao
 from rdr_service.model.ppsc import Participant, Site, NPHOptInEvent, ProfileUpdatesEvent, ParticipantStatusEvent, \
     ConsentEvent
+from rdr_service.model.rex import ParticipantMapping
 from rdr_service.model.study_nph import EligibleParticipants
 
 
@@ -26,6 +27,10 @@ def get_participant_by_biobank_id(self, *, biobank_id: int):
         with self.session() as session:
             return session.query(Participant).filter(Participant.biobank_id == biobank_id).all()
 
+    def get_all_participants_from_list(self, *, participant_ids: List[int]):
+        with self.session() as session:
+            return session.query(Participant).filter(Participant.id.in_(participant_ids)).all()
+
 
 class SiteDao(UpsertableDao):
 
@@ -151,6 +156,9 @@ def get_eligible_participant_records(self):
                     ParticipantStatusEvent.participant_id == ProfileUpdatesEvent.participant_id,
                     ParticipantStatusEvent.event_type_name.ilike('%Test Account%')
                 )
+            ).outerjoin(
+                ParticipantMapping,
+                ParticipantMapping.primary_participant_id == ProfileUpdatesEvent.participant_id
             ).filter(
                 ProfileUpdatesEvent.data_element_name.in_(
                     ['piiname_first',
@@ -162,7 +170,8 @@ def get_eligible_participant_records(self):
                 ),
                 profile_updates_alias.id.is_(None),
                 EligibleParticipants.id.is_(None),
-                ParticipantStatusEvent.id.is_(None)
+                ParticipantStatusEvent.id.is_(None),
+                ParticipantMapping.id.is_(None)
             ).group_by(
                 ProfileUpdatesEvent.participant_id,
                 case(

rdr_service/ppsc/ppsc_partner_data_sync.py

@@ -1,10 +1,11 @@
 import logging
+from sqlalchemy.exc import IntegrityError
 
 from rdr_service.config import NPH_STUDY_ID, AOU_STUDY_ID
-from rdr_service.dao.ppsc_dao import PPSCNphOptEventInDao
+from rdr_service.dao.ppsc_dao import PPSCNphOptEventInDao, ParticipantDao
 from rdr_service.dao.ppsc_partner_transfer_dao import RTIDataTransferBaseDao
 from rdr_service.dao.rex_dao import RexParticipantMappingDao
-from rdr_service.dao.study_nph_dao import EligibleParticipantsDao
+from rdr_service.dao.study_nph_dao import EligibleParticipantsDao, NphParticipantDao
 from rdr_service.model.ppsc_partner_data_transfer import RTINphOptIn
 
 
@@ -15,6 +16,8 @@ def __init__(self):
         self.eligible_dao = EligibleParticipantsDao()
         self.nph_opt_in_event_dao = PPSCNphOptEventInDao()
         self.rex_mapping_dao = RexParticipantMappingDao()
+        self.nph_participant_dao = NphParticipantDao()
+        self.ppsc_participant_dao = ParticipantDao()
         self.usable_nph_objects = None
         self.items_ready_for_sync = []
 
@@ -48,8 +51,12 @@ def get_language_pref(cls, value: str) -> int:
 
     def sync_items(self):
         self.get_items_for_sync()
+        all_participant_data = self.ppsc_participant_dao.get_all_participants_from_list(
+            participant_ids=[obj.participant_id for obj in self.items_ready_for_sync])
+
         for item in self.items_ready_for_sync:
             usable_nph_obj = self.get_nph_obj_from_list()
+            participant_data = [obj for obj in all_participant_data if obj.id == item.participant_id]
             self.nph_opt_in_dao.insert(self.nph_opt_in_dao.model_type(**{
                 'nph_participant_id': usable_nph_obj.participant_id,
                 'first_name': item.first_name,
@@ -71,6 +78,14 @@ def sync_items(self):
                 'primary_study_id': AOU_STUDY_ID
 
             }))
+            try:
+                self.nph_participant_dao.insert(self.nph_participant_dao.model_type(**{
+                    'id': usable_nph_obj.participant_id,
+                    'biobank_id': participant_data[0].biobank_id,
+                }))
+            except IntegrityError as e:
+                logging.error(f'The NPH participant record already exists for {usable_nph_obj.participant_id}: {e}')
+
             self.usable_nph_objects.pop(0)
 
     def run_sync(self):

tests/ppsc_tests/test_ppsc_partner_data_sync.py

@@ -6,7 +6,7 @@
 from rdr_service.dao.ppsc_dao import ParticipantDao
 from rdr_service.dao.ppsc_partner_transfer_dao import RTIDataTransferBaseDao
 from rdr_service.dao.rex_dao import RexStudyDao, RexParticipantMappingDao
-from rdr_service.dao.study_nph_dao import EligibleParticipantsDao
+from rdr_service.dao.study_nph_dao import EligibleParticipantsDao, NphParticipantDao
 from rdr_service.data_gen.generators.nph import NphDataGenerator
 from rdr_service.data_gen.generators.ppsc import PPSCDataGenerator
 from rdr_service.model.ppsc_partner_data_transfer import RTINphOptIn
@@ -24,6 +24,7 @@ def setUp(self):
         self.nph_opt_in_dao = RTIDataTransferBaseDao(RTINphOptIn)
         self.rex_study_dao = RexStudyDao()
         self.rex_mapping_dao = RexParticipantMappingDao()
+        self.nph_participant_dao = NphParticipantDao()
         self.faker = Faker()
 
         activities = [
@@ -152,7 +153,6 @@ def test_get_eligible_nph_participants(self) -> None:
 
         # nph opt in records
         current_opt_in_records = [obj for obj in self.nph_opt_in_dao.get_all()]
-
         self.assertEqual(len(current_opt_in_records), len(nph_opt_in_sync.items_ready_for_sync))
         self.assertTrue(all(obj.first_name is not None for obj in current_opt_in_records))
         self.assertTrue(all(obj.last_name is not None for obj in current_opt_in_records))
@@ -166,10 +166,16 @@ def test_get_eligible_nph_participants(self) -> None:
         current_mappings = self.rex_mapping_dao.get_all()
         self.assertEqual(len(current_mappings), len(nph_opt_in_sync.items_ready_for_sync))
 
+        # check NPH record was created
+        current_nph_participants = self.nph_participant_dao.get_all()
+        synced_nph_id = current_opt_in_records[0].nph_participant_id
+        self.assertTrue(synced_nph_id in [obj.id for obj in current_nph_participants])
+
     def tearDown(self):
         super().tearDown()
         self.clear_table_after_test("ppsc.activity")
         self.clear_table_after_test("ppsc.participant")
         self.clear_table_after_test("ppsc.participant_event_activity")
         self.clear_table_after_test("ppsc.profile_updates_event")
         self.clear_table_after_test("ppsc.nph_opt_in_event")
+        self.clear_table_after_test("nph.participant")