process_tracker_python-143 Add insert/update counts for process_tracking

✨ Added ability to break out record counts by processing type (inserts, updates)

Instead of just having an overall total record count for process runs, support is added for breaking out the total count by inserts and updates for a given run.
Totals are still available, but this allows for finer understanding of what actually was done during the process run.

Closes #143

Author: Alex Meadows

dbscripts/mysql_process_tracker.sql

@@ -338,6 +338,8 @@ create table process_tracking
 	process_run_actor_id int null,
 	is_latest_run tinyint(1) not null,
 	process_run_name varchar(250) null,
+	process_run_insert_count int default 0 not null,
+	process_run_update_count int default 0 not null,
 	constraint process_tracking_process_run_name_uindex
 		unique (process_run_name),
 	constraint process_tracking_ibfk_1

dbscripts/postgresql_process_tracker.sql

@@ -365,7 +365,9 @@ create table process_tracking
 		constraint process_tracking_fk03
 			references actor_lkup,
 	is_latest_run boolean default false,
-	process_run_name varchar(250) null
+	process_run_name varchar(250) null,
+	process_run_insert_count int default 0 not null,
+	process_run_update_count int default 0 not null
 );
 
 comment on table process_tracking is 'Tracking table of process runs.';

process_tracker/models/process.py

@@ -508,6 +508,8 @@ class ProcessTracking(Base):
     process_run_start_date_time = Column(DateTime, nullable=False)
     process_run_end_date_time = Column(DateTime, nullable=True)
     process_run_record_count = Column(Integer, nullable=False, default=0)
+    process_run_insert_count = Column(Integer, nullable=False, default=0)
+    process_run_update_count = Column(Integer, nullable=False, default=0)
     process_run_actor_id = Column(
         Integer, ForeignKey("process_tracker.actor_lkup.actor_id"), nullable=False
     )

process_tracker/process_tracker.py

@@ -1323,22 +1323,53 @@ def set_process_run_low_high_dates(self, low_date=None, high_date=None):
 
         self.session.commit()
 
-    def set_process_run_record_count(self, num_records):
+    def record_count_manager(self, original_count, num_records):
+        """
+        Given two record counts, one the overall count and the other the current count, process and return the adjusted
+        amount.
+        :param original_count: The original overall count of records
+        :type original_count: int
+        :param num_records: The adjusted count of records
+        :type num_records: int
+        """
+        if original_count == 0 or original_count is None:
+            return_count = num_records
+        else:
+            return_count = original_count + (num_records - original_count)
+
+        return return_count
+
+    def set_process_run_record_count(self, num_records, processing_type=None):
         """
         For the given process run, set the process_run_record_count for the number of records processed.  Will also
         update the process' total_record_count - the total number of records ever processed by that process
-        :param num_records:
+        :param num_records: Count of number of records processed.
+        :type num_records: int
+        :param processing_type: Type of records being processed.  Valid values: None, insert, update.  None will only
+                                update overall counts
         :return:
         """
         process_run_records = self.process.total_record_count
+        process_run_inserts = self.process_tracking_run.process_run_insert_count
+        process_run_updates = self.process_tracking_run.process_run_update_count
 
-        if process_run_records == 0 or process_run_records is None:
-
-            self.process.total_record_count = num_records
-        else:
-            self.process.total_record_count = self.process.total_record_count + (
-                num_records - process_run_records
+        if processing_type == "insert":
+            self.process_tracking_run.process_run_insert_count = self.record_count_manager(
+                original_count=process_run_inserts, num_records=num_records
+            )
+        elif processing_type == "update":
+            self.process_tracking_run.process_run_update_count = self.record_count_manager(
+                original_count=process_run_updates, num_records=num_records
             )
+        elif processing_type is not None:
+            error_msg = "Processing type not recognized."
+            self.logger.error(error_msg)
+            raise Exception(error_msg)
+
+        self.process.total_record_count = self.record_count_manager(
+            original_count=process_run_records, num_records=num_records
+        )
 
         self.process_tracking_run.process_run_record_count = num_records
+
         self.session.commit()

tests/test_process_tracker.py

@@ -1877,6 +1877,80 @@ def test_set_process_run_record_count_twice(self):
 
         self.assertEqual(expected_result, given_result)
 
+    def test_set_process_run_record_count_insert(self):
+        """
+        Testing that if record counts are provided for a given process_run, and the processing type is insert,
+        update the insert count for the process run.
+        """
+        initial_record_count = 1000
+
+        self.process_tracker.set_process_run_record_count(
+            num_records=initial_record_count, processing_type="insert"
+        )
+
+        given_count = self.process_tracker.process_tracking_run.process_run_insert_count
+        expected_count = 1000
+
+        self.assertEqual(expected_count, given_count)
+
+    def test_set_process_run_record_count_insert_twice(self):
+        """
+        Testing that if record counts are provided for a given process_run multiple times, and the processing type is insert,
+        update the insert count for the process run.
+        """
+        initial_record_count = 1000
+        modified_record_count = 1500
+
+        self.process_tracker.set_process_run_record_count(
+            num_records=initial_record_count, processing_type="insert"
+        )
+
+        self.process_tracker.set_process_run_record_count(
+            num_records=modified_record_count, processing_type="insert"
+        )
+
+        given_count = self.process_tracker.process_tracking_run.process_run_insert_count
+        expected_count = 1500
+
+        self.assertEqual(expected_count, given_count)
+
+    def test_set_process_run_record_count_update(self):
+        """
+        Testing that if record counts are provided for a given process_run, and the processing type is update,
+        update the update count for the process run.
+        """
+        initial_record_count = 1000
+
+        self.process_tracker.set_process_run_record_count(
+            num_records=initial_record_count, processing_type="update"
+        )
+
+        given_count = self.process_tracker.process_tracking_run.process_run_update_count
+        expected_count = 1000
+
+        self.assertEqual(expected_count, given_count)
+
+    def test_set_process_run_record_count_update_twice(self):
+        """
+        Testing that if record counts are provided for a given process_run multiple times, and the processing type is update,
+        update the update count for the process run.
+        """
+        initial_record_count = 1000
+        modified_record_count = 1500
+
+        self.process_tracker.set_process_run_record_count(
+            num_records=initial_record_count, processing_type="update"
+        )
+
+        self.process_tracker.set_process_run_record_count(
+            num_records=modified_record_count, processing_type="update"
+        )
+
+        given_count = self.process_tracker.process_tracking_run.process_run_update_count
+        expected_count = 1500
+
+        self.assertEqual(expected_count, given_count)
+
     def test_register_source_dataset_type(self):
         """
         When both a source and dataset_type are provided, the source is registered to the dataset_type.