🗃️ Add more robust capabilities to extract_tracking and related tables.

Author: OpenDataAlex

models/extract.py

@@ -7,6 +7,16 @@
 from models.model_base import Base
 
 
+class ExtractStatus(Base):
+
+    __tablename__ = "extract_status_lkup"
+
+    extract_status_id = Column(Integer, Sequence('extract_status_lkup_status_id_seq'), primary_key=True)
+    extract_status_name = Column(String(75), nullable=False, unique=True)
+
+    extracts = relationship("Extract")
+
+
 class Extract(Base):
 
     __tablename__ = "extract_tracking"
@@ -15,6 +25,10 @@ class Extract(Base):
     extract_source_id = Column(Integer, ForeignKey("source_lkup.source_id"))
     extract_filename = Column(String(750), nullable=False, unique=True)
     extract_location_id = Column(Integer, ForeignKey('location_lkup.location_id'))
+    extract_process_run_id = Column(Integer, ForeignKey('process_tracking.process_tracking_id'))
+    extract_status_id = Column(Integer, ForeignKey('extract_status_lkup.extract_status_id'))
+
+    process_tracking = relationship("ProcessTracking")
 
 
 class Location(Base):

models/process.py

@@ -2,10 +2,11 @@
 # Models for Process entities
 
 
-from sqlalchemy import Column, DateTime, ForeignKey, Integer, Sequence, String
+from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, Sequence, String
 from sqlalchemy.orm import relationship
 
 from models.model_base import default_date, Base
+from models.extract import Extract
 
 
 class ErrorType(Base):
@@ -132,9 +133,11 @@ class ProcessTracking(Base):
     process_run_end_date_time = Column(DateTime, nullable=True)
     process_run_record_count = Column(Integer, nullable=False, default=0)
     process_run_actor_id = Column(Integer, ForeignKey('actor_lkup.actor_id'))
+    is_latest_run = Column(Boolean, nullable=False, default=False)
 
-    process = relationship("Process", back_populates="process_tracking")
     errors = relationship("ErrorTracking", back_populates="error_tracking")
+    extracts = relationship("Extract", back_populates="process_tracking")
+    process = relationship("Process", back_populates="process_tracking")
 
     def __repr__(self):
 

process_tracker/extract_tracking.py

@@ -0,0 +1,22 @@
+# Extract Tracking
+# Used in the creation and editing of extract records.  Used in conjunction with process tracking.
+
+from models.extract import Extract, Location
+from models.source import Source
+
+
+class ExtractTracker:
+
+    def __init__(self, source_name, filename, location, process_name):
+        """
+        ExtractTracker is the primary engine for tracking data extracts
+        :param source_name: Name of the source where data extract is from.
+        :type source_name: string
+        :param filename: Name of the data extract file.
+        :type filename:  string
+        :param location: Location (filepath, s3 bucket, etc.) where the file is stored
+        :type location: string
+        :param process_name: Name of the process that produced the data extract.
+        :type process_name: string
+        """
+

process_tracker/process.py

@@ -1,3 +1,6 @@
+# Process Tracking
+# Used in the creation and editing of process tracking records.
+
 from datetime import datetime
 import logging
 
@@ -13,7 +16,7 @@ class ProcessTracker:
 
     def __init__(self, process_name, process_type, actor_name, tool_name, source_name):
         """
-        ProcessTracker is the primary enginer for tracking data integration processes.
+        ProcessTracker is the primary engine for tracking data integration processes.
         :param process_name: Name of the process being tracked.
         :param actor_name: Name of the person or environment runnning the process.
         :param tool_name: Name of the tool used to run the process.
@@ -58,6 +61,7 @@ def register_new_process_run(self):
             # Must validate that the process is not currently running.
 
             if last_run.process_status_id != self.process_status_running:
+                last_run.is_latest_run = False
                 new_run_flag = True
                 new_run_id = last_run.process_run_id + 1
             else:
@@ -68,7 +72,8 @@ def register_new_process_run(self):
                                       , process_status_id=self.process_status_running
                                       , process_run_id=new_run_id
                                       , process_run_start_date_time=datetime.now()
-                                      , process_run_actor_id=self.actor.actor_id)
+                                      , process_run_actor_id=self.actor.actor_id
+                                      , is_latest_run = True)
 
             session.add(new_run)
             session.commit()

process_tracker/process_tracking.py

@@ -15,10 +15,10 @@ class TestProcessTracking(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.process_tracker = ProcessTracker(process_name='Testing Process Tracking Initialization'
-                                         , process_type='Extract'
-                                         , actor_name='UnitTesting'
-                                         , tool_name='Spark'
-                                         , source_name='Unittests')
+                                             , process_type='Extract'
+                                             , actor_name='UnitTesting'
+                                             , tool_name='Spark'
+                                             , source_name='Unittests')
 
     @classmethod
     def tearDownClass(cls):
@@ -93,6 +93,25 @@ def test_register_new_process_run_exception(self):
         return self.assertTrue('The process Testing Process Tracking Initialization '
                                'is currently running.' in str(context.exception))
 
+    def test_register_new_process_run_with_previous_run(self):
+        """
+        Testing that a new run record is created if there is another instance of same process in 'completed' or 'failed'
+        status.  Also flips the is_latest_run flag on previous run to False.
+        :return:
+        """
+
+        self.process_tracker.change_run_status(new_status='completed')
+        self.process_tracker.register_new_process_run()
+
+        process_runs = session.query(ProcessTracking)\
+                              .filter(ProcessTracking.process_id == self.process_id)\
+                              .order_by(ProcessTracking.process_tracking_id)
+
+        given_result = process_runs[0].is_latest_run
+        expected_result = False
+
+        self.assertEqual(given_result, expected_result)
+
     def test_change_run_status_complete(self):
         """
         Testing that when changing the run status from 'running' to 'complete' the run record updates successfully.