process_tracker_python-90 Add contact information to data source

✨ Added Contact

Author: Alex Meadows

.gitignore

@@ -175,3 +175,4 @@ pip-selfcheck.json
 .idea/process_tracker_python.iml
 /tests/s3:/
 /tests/s3:\/
+/.idea/inspectionProfiles/

dbscripts/mysql_process_tracker.sql

@@ -1,5 +1,17 @@
 USE process_tracker;
 
+create table contact_lkup
+(
+	contact_id int auto_increment
+		primary key,
+	contact_name varchar(250) not null,
+	contact_email varchar(750) null,
+	constraint contact_lkup_contact_email_uindex
+		unique (contact_email),
+	constraint contact_lkup_contact_name_uindex
+		unique (contact_name)
+);
+
 create table dataset_type_lkup
 (
 	dataset_type_id int auto_increment,
@@ -132,6 +144,17 @@ create table source_lkup
 		unique (source_name)
 );
 
+create table process_tracker.source_contact
+(
+	source_id int not null,
+	contact_id int not null,
+	primary key (source_id, contact_id),
+	constraint source_contact_fk01
+		foreign key (source_id) references process_tracker.source_lkup (source_id),
+	constraint source_contact_fk02
+		foreign key (contact_id) references process_tracker.contact_lkup (contact_id)
+);
+
 create table system_lkup
 (
 	system_id int auto_increment

dbscripts/postgresql_process_tracker.sql

@@ -4,6 +4,23 @@ create schema process_tracker;
 
 alter schema process_tracker owner to pt_admin;
 
+create table process_tracker.contact_lkup
+(
+	contact_id serial not null
+		constraint contact_lkup_pk
+			primary key,
+	contact_name varchar(250) not null,
+	contact_email varchar(750)
+);
+
+alter table process_tracker.contact_lkup owner to pt_admin;
+
+create unique index contact_lkup_contact_email_uindex
+	on process_tracker.contact_lkup (contact_email);
+
+create unique index contact_lkup_contact_name_uindex
+	on process_tracker.contact_lkup (contact_name);
+
 create table process_tracker.dataset_type_lkup
 (
 	dataset_type_id serial not null
@@ -91,6 +108,20 @@ alter table source_lkup owner to pt_admin;
 create unique index source_lkup_udx01
 	on source_lkup (source_name);
 
+create table process_tracker.source_contact
+(
+	source_id integer not null
+		constraint source_contact_fk01
+			references process_tracker.source_lkup,
+	contact_id integer not null
+		constraint source_contact_fk02
+			references process_tracker.contact_lkup,
+	constraint source_contact_pk
+		primary key (source_id, contact_id)
+);
+
+alter table process_tracker.source_contact owner to pt_admin;
+
 create table process_status_lkup
 (
 	process_status_id serial not null

process_tracker/process_tracker.py

@@ -16,6 +16,7 @@
 from process_tracker.utilities import utilities
 
 from process_tracker.models.actor import Actor
+from process_tracker.models.contact import Contact
 from process_tracker.models.extract import (
     Extract,
     ExtractProcess,
@@ -39,6 +40,7 @@
 from process_tracker.models.source import (
     DatasetType,
     Source,
+    SourceContact,
     SourceDatasetType,
     SourceObject,
     SourceObjectDatasetType,
@@ -399,6 +401,37 @@ def find_extracts_by_process(self, extract_process_name, status="ready"):
 
         return process_files
 
+    def find_process_contacts(self, process):
+        """
+        Based on the process, find the process' contacts as well as the contacts for any sources used and return them
+        as a list.
+        :param process: The process' process_id
+        :type process: int
+        :return:
+        """
+        contacts = list()
+
+        result = (
+            self.session.query(Contact.contact_name, Contact.contact_email)
+            .join(SourceContact)
+            .join(Source)
+            .join(ProcessSource)
+            .join(Process)
+            .join(ProcessTracking)
+            .filter(ProcessTracking.process_id == process)
+        )
+
+        for contact in result:
+            contacts.append(
+                {
+                    "contact_name": contact.contact_name,
+                    "contact_email": contact.contact_email,
+                    "contact_type": "source",
+                }
+            )
+
+        return contacts
+
     def get_latest_tracking_record(self, process):
         """
         For the given process, find the latest tracking record.

tests/test_process_tracker.py

@@ -13,6 +13,7 @@
 from moto import mock_s3
 from sqlalchemy.orm import aliased, Session
 
+from process_tracker.models.contact import Contact
 from process_tracker.models.extract import (
     Extract,
     ExtractDatasetType,
@@ -35,19 +36,21 @@
 from process_tracker.models.source import (
     DatasetType,
     Source,
+    SourceContact,
     SourceDatasetType,
     SourceObject,
     SourceObjectDatasetType,
 )
 
-from process_tracker.utilities.data_store import DataStore
+from process_tracker.utilities.data_store import DataStore, ClusterProcess
 from process_tracker.extract_tracker import ExtractTracker
 from process_tracker.process_tracker import ProcessTracker
 from process_tracker.utilities import utilities
 
 test_bucket = "test_bucket"
 
 
+
 # @mock_s3
 class TestProcessTracker(unittest.TestCase):
     @classmethod
@@ -64,6 +67,7 @@ def setUpClass(cls):
 
     @classmethod
     def tearDownClass(cls):
+        cls.session.query(ClusterProcess).delete()
         cls.session.query(Location).delete()
         cls.session.query(DatasetType).delete()
         cls.session.query(ProcessSourceObject).delete()
@@ -656,6 +660,39 @@ def test_find_extracts_by_process_not_descending(self):
 
         self.assertNotEqual(expected_result, given_result)
 
+    def test_find_process_contacts(self):
+        """
+        Testing that when passed a process_id, a list of source contacts will be returned.
+        :return:
+        """
+        contact = DataStore().get_or_create_item(
+            model=Contact,
+            contact_name="Test Contact",
+            contact_email="testcontact@test.com",
+        )
+
+        source = DataStore().get_or_create_item(model=Source, source_name="Unittests")
+
+        DataStore().get_or_create_item(
+            model=SourceContact,
+            source_id=source.source_id,
+            contact_id=contact.contact_id,
+        )
+
+        given_result = self.process_tracker.find_process_contacts(
+            process=self.process_id
+        )
+
+        expected_result = [
+            {
+                "contact_name": "Test Contact",
+                "contact_email": "testcontact@test.com",
+                "contact_type": "source",
+            }
+        ]
+
+        self.assertEqual(expected_result, given_result)
+
     def test_initializing_process_tracking(self):
         """
         Testing that when ProcessTracking is initialized, the necessary objects are created.