process_tracker_python-9 Process can have more than one source

✨ Processes now fail on dependency running or failed.

Dependencies of a given process will now halt the process from starting
if they are currently running or failed their last run.

Closes: #6

Author: OpenDataAlex

process_tracker/process_tracker.py

@@ -7,14 +7,15 @@
 from os.path import join
 
 import boto3
+from sqlalchemy.orm import aliased
 
 from process_tracker.data_store import DataStore
 from process_tracker.extract_tracker import ExtractTracker
 from process_tracker.location_tracker import LocationTracker
 
 from process_tracker.models.actor import Actor
 from process_tracker.models.extract import Extract, ExtractProcess, ExtractStatus, Location
-from process_tracker.models.process import ErrorTracking, ErrorType, Process, ProcessTracking, ProcessStatus, ProcessSource, ProcessTarget, ProcessType
+from process_tracker.models.process import ErrorTracking, ErrorType, Process, ProcessDependency, ProcessTracking, ProcessStatus, ProcessSource, ProcessTarget, ProcessType
 from process_tracker.models.source import Source
 from process_tracker.models.tool import Tool
 
@@ -257,12 +258,28 @@ def register_new_process_run(self):
         When a new process instance is starting, register the run in process tracking.
         :return:
         """
+        child_process = aliased(Process)
+        parent_process = aliased(Process)
 
         last_run = self.get_latest_tracking_record(process=self.process)
 
         new_run_flag = True
         new_run_id = 1
 
+        # Need to check the status of any dependencies.  If dependencies are running or failed, halt this process.
+
+        dependency_hold = self.session.query(ProcessDependency)\
+                                      .join(parent_process, ProcessDependency.parent_process_id == parent_process.process_id)\
+                                      .join(child_process, ProcessDependency.child_process_id == child_process.process_id)\
+                                      .join(ProcessTracking, ProcessTracking.process_id == parent_process.process_id) \
+                                      .join(ProcessStatus, ProcessStatus.process_status_id == ProcessTracking.process_status_id) \
+                                      .filter(child_process.process_id == self.process.process_id) \
+                                      .filter(ProcessStatus.process_status_name.in_(('running', 'failed'))) \
+                                      .count()
+
+        if dependency_hold > 0:
+            raise Exception('Processes that this process is dependent on are running or failed.')
+
         if last_run:
             # Must validate that the process is not currently running.
 

tests/test_extract_tracker.py

@@ -3,7 +3,7 @@
 import unittest
 
 from process_tracker.models.extract import Extract, ExtractProcess, Location
-from process_tracker.models.process import Process, ProcessSource, ProcessTarget, ProcessTracking
+from process_tracker.models.process import Process, ProcessDependency, ProcessSource, ProcessTarget, ProcessTracking
 
 from process_tracker.data_store import DataStore
 from process_tracker.extract_tracker import ExtractTracker
@@ -32,6 +32,7 @@ def tearDownClass(cls):
         cls.session.query(ProcessTracking).delete()
         cls.session.query(ProcessSource).delete()
         cls.session.query(ProcessTarget).delete()
+        cls.session.query(ProcessDependency).delete()
         cls.session.query(Process).delete()
         cls.session.commit()
 

tests/test_process_tracker.py

@@ -8,7 +8,7 @@
 from sqlalchemy.orm import aliased, Session
 
 from process_tracker.models.extract import Extract, ExtractProcess, ExtractStatus, Location
-from process_tracker.models.process import ErrorType, ErrorTracking, Process, ProcessSource, ProcessTarget, ProcessTracking
+from process_tracker.models.process import ErrorType, ErrorTracking, Process, ProcessDependency, ProcessSource, ProcessTarget, ProcessTracking
 
 from process_tracker.data_store import DataStore
 from process_tracker.extract_tracker import ExtractTracker
@@ -19,15 +19,16 @@ class TestProcessTracking(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
-        data_store = DataStore()
-        cls.session = data_store.session
-        cls.data_store_type = data_store.data_store_type
+        cls.data_store = DataStore()
+        cls.session = cls.data_store.session
+        cls.data_store_type = cls.data_store.data_store_type
 
     @classmethod
     def tearDownClass(cls):
         cls.session.query(Location).delete()
         cls.session.query(ProcessSource).delete()
         cls.session.query(ProcessTarget).delete()
+        cls.session.query(ProcessDependency).delete()
         cls.session.query(Process).delete()
         cls.session.commit()
 
@@ -400,6 +401,80 @@ def test_register_new_process_run_with_previous_run(self):
 
         self.assertEqual(expected_result, given_result)
 
+    def test_register_new_process_run_dependencies_completed(self):
+        """
+        Testing that for a given process, if there are completed dependencies, then the process run is created.
+        :return:
+        """
+        dependent_process = ProcessTracker(process_name='Testing Process Tracking Dependency'
+                                           , process_type='Extract'
+                                           , actor_name='UnitTesting'
+                                           , tool_name='Spark'
+                                           , sources='Unittests'
+                                           , targets='Unittests')
+
+        dependent_process.change_run_status(new_status='completed')
+        self.process_tracker.change_run_status(new_status='completed')
+        self.data_store.get_or_create(model=ProcessDependency
+                                      , parent_process_id=dependent_process.process_tracking_run.process_id
+                                      , child_process_id=self.process_id)
+
+        self.process_tracker.register_new_process_run()
+
+        given_count = self.session.query(ProcessTracking) \
+                                  .filter(ProcessTracking.process_id == self.process_id) \
+                                  .count()
+
+        expected_count = 2
+
+        self.assertEqual(expected_count, given_count)
+
+    def test_register_new_process_run_dependencies_running(self):
+        """
+        Testing that for a given process, if there are running dependencies, then the process run is prevented from starting.
+        :return:
+        """
+        dependent_process = ProcessTracker(process_name='Testing Process Tracking Dependency Running'
+                                           , process_type='Extract'
+                                           , actor_name='UnitTesting'
+                                           , tool_name='Spark'
+                                           , sources='Unittests'
+                                           , targets='Unittests')
+
+        dependent_process.change_run_status(new_status='running')
+        self.process_tracker.change_run_status(new_status='completed')
+        self.data_store.get_or_create(model=ProcessDependency
+                                      , parent_process_id=dependent_process.process_tracking_run.process_id
+                                      , child_process_id=self.process_id)
+
+        with self.assertRaises(Exception) as context:
+            self.process_tracker.register_new_process_run()
+
+        return self.assertTrue('Processes that this process is dependent on are running or failed.' in str(context.exception))
+
+    def test_register_new_process_run_dependencies_failed(self):
+        """
+        Testing that for a given process, if there are failed dependencies, then the process run is prevented from starting.
+        :return:
+        """
+        dependent_process = ProcessTracker(process_name='Testing Process Tracking Dependency Failed'
+                                           , process_type='Extract'
+                                           , actor_name='UnitTesting'
+                                           , tool_name='Spark'
+                                           , sources='Unittests'
+                                           , targets='Unittests')
+
+        dependent_process.change_run_status(new_status='failed')
+        self.process_tracker.change_run_status(new_status='completed')
+        self.data_store.get_or_create(model=ProcessDependency
+                                      , parent_process_id=dependent_process.process_tracking_run.process_id
+                                      , child_process_id=self.process_id)
+
+        with self.assertRaises(Exception) as context:
+            self.process_tracker.register_new_process_run()
+
+        return self.assertTrue('Processes that this process is dependent on are running or failed.' in str(context.exception))
+
     def test_register_process_sources_one_source(self):
         """
         Testing that when a new process is registered, a source registered as well.